vamossagar12 commented on code in PR #14372:
URL: https://github.com/apache/kafka/pull/14372#discussion_r1670233898
##########
connect/runtime/src/test/java/org/apache/kafka/connect/util/KafkaBasedLogTest.java:
##########
@@ -403,6 +404,54 @@ public void testGetOffsetsConsumerErrorOnReadToEnd()
throws Exception {
verifyStartAndStop();
}
+ @Test
+ public void testOffsetReadFailureWhenWorkThreadFails() throws Exception {
+ RuntimeException exception = new RuntimeException();
+ Set<TopicPartition> tps = new HashSet<>(Arrays.asList(TP0, TP1));
+ Map<TopicPartition, Long> endOffsets = new HashMap<>();
+ endOffsets.put(TP0, 0L);
+ endOffsets.put(TP1, 0L);
+ admin = mock(TopicAdmin.class);
+ when(admin.endOffsets(eq(tps)))
+ .thenReturn(endOffsets)
+ .thenThrow(exception)
+ .thenReturn(endOffsets);
+
+ store.start();
+
+ AtomicInteger numSuccesses = new AtomicInteger();
+ AtomicInteger numFailures = new AtomicInteger();
+ AtomicReference<FutureCallback<Void>> finalSuccessCallbackRef = new
AtomicReference<>();
+ final FutureCallback<Void> successCallback = new
FutureCallback<>((error, result) -> numSuccesses.getAndIncrement());
+ final FutureCallback<Void> firstFailedCallback = new
FutureCallback<>((error, result) -> {
+ numFailures.getAndIncrement();
+ // We issue another readToEnd call here to simulate the case that
more read requests can come in while
+ // the failure is being handled in the WorkThread.
+ final FutureCallback<Void> finalSuccessCallback = new
FutureCallback<>((e, r) -> numSuccesses.getAndIncrement());
+ finalSuccessCallbackRef.set(finalSuccessCallback);
+ store.readToEnd(finalSuccessCallback);
+ });
+ final FutureCallback<Void> subsequentFailedCallback = new
FutureCallback<>((error, result) -> numFailures.getAndIncrement());
+
+ store.readToEnd(successCallback);
+ store.readToEnd(firstFailedCallback);
+ store.readToEnd(subsequentFailedCallback);
Review Comment:
yeah that makes sense. I updated the test based on your suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
