apoorvmittal10 commented on code in PR #22881:
URL: https://github.com/apache/kafka/pull/22881#discussion_r3615089274
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -2505,6 +2856,71 @@ public void testAsyncReadExceptionReleasesLocks() {
Mockito.verify(sp0).releaseFetchLock(fetchId);
}
+ /**
+ * Tests that when the async read fails and the error handling in
completeShareFetchAsyncRequest
+ * itself throws (here the exception handler callback), the partition
locks are still released
+ * This guards the try-finally around the metric recording and exception
handling in the get()-failure
+ * path.
+ */
+ @Test
+ public void
testCompleteShareFetchAsyncRequestReleasesLocksWhenExceptionHandlingThrows() {
+ String groupId = "grp";
+ Uuid topicId = Uuid.randomUuid();
+ LogReader logReader = mock(LogReader.class);
+ PartitionMetadataProvider metadataProvider =
mock(PartitionMetadataProvider.class);
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(topicId, new
TopicPartition("foo", 0));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new
LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+
+ CompletableFuture<Map<TopicIdPartition,
ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId,
Uuid.randomUuid().toString(),
+ future, List.of(tp0), BATCH_OPTIMIZED, BATCH_SIZE,
MAX_FETCH_RECORDS, BROKER_TOPIC_STATS);
+
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
asyncReadFuture = new CompletableFuture<>();
+ when(logReader.readAsync(any(), any(), any(), any(),
anyBoolean())).thenReturn(asyncReadFuture);
+
+ when(sp0.maybeAcquireFetchLock(any())).thenReturn(true);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp0.nextFetchOffset()).thenReturn(0L);
+ when(sp0.fetchOffsetMetadata(anyLong()))
+ .thenReturn(Optional.empty())
+ .thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(metadataProvider.endOffsetMetadata(any(), any())).thenReturn(new
LogOffsetMetadata(1, 1, 1));
+
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy =
mockPartitionMaxBytes(Set.of(tp0));
+
+ // The exception handler throws while handling the async read failure,
simulating a failure
+ // inside the catch block's error handling. Locks must still be
released via the finally.
Review Comment:
Done.
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -2505,6 +2856,71 @@ public void testAsyncReadExceptionReleasesLocks() {
Mockito.verify(sp0).releaseFetchLock(fetchId);
}
+ /**
+ * Tests that when the async read fails and the error handling in
completeShareFetchAsyncRequest
+ * itself throws (here the exception handler callback), the partition
locks are still released
+ * This guards the try-finally around the metric recording and exception
handling in the get()-failure
+ * path.
+ */
+ @Test
+ public void
testCompleteShareFetchAsyncRequestReleasesLocksWhenExceptionHandlingThrows() {
+ String groupId = "grp";
+ Uuid topicId = Uuid.randomUuid();
+ LogReader logReader = mock(LogReader.class);
+ PartitionMetadataProvider metadataProvider =
mock(PartitionMetadataProvider.class);
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(topicId, new
TopicPartition("foo", 0));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new
LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+
+ CompletableFuture<Map<TopicIdPartition,
ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId,
Uuid.randomUuid().toString(),
+ future, List.of(tp0), BATCH_OPTIMIZED, BATCH_SIZE,
MAX_FETCH_RECORDS, BROKER_TOPIC_STATS);
+
+ CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>>
asyncReadFuture = new CompletableFuture<>();
+ when(logReader.readAsync(any(), any(), any(), any(),
anyBoolean())).thenReturn(asyncReadFuture);
+
+ when(sp0.maybeAcquireFetchLock(any())).thenReturn(true);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp0.nextFetchOffset()).thenReturn(0L);
+ when(sp0.fetchOffsetMetadata(anyLong()))
+ .thenReturn(Optional.empty())
+ .thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(metadataProvider.endOffsetMetadata(any(), any())).thenReturn(new
LogOffsetMetadata(1, 1, 1));
+
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy =
mockPartitionMaxBytes(Set.of(tp0));
+
+ // The exception handler throws while handling the async read failure,
simulating a failure
+ // inside the catch block's error handling. Locks must still be
released via the finally.
+ BiConsumer<SharePartitionKey, Throwable> exceptionHandler =
mockExceptionHandler();
+ doThrow(new RuntimeException("handler
boom")).when(exceptionHandler).accept(any(), any());
+
+ Uuid fetchId = Uuid.randomUuid();
+ DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withReplicaManager(replicaManager)
+ .withLogReader(logReader)
+ .withMetadataProvider(metadataProvider)
+ .withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
+ .withSharePartitions(sharePartitions)
+ .withExceptionHandler(exceptionHandler)
+ .withFetchId(fetchId)
+ .build();
+
+ // First tryComplete triggers the async read, which stays pending
(slow path).
+ assertFalse(delayedShareFetch.tryComplete());
+ assertNotNull(delayedShareFetch.pendingFetch());
+ // Complete the async read exceptionally.
+ asyncReadFuture.completeExceptionally(new RuntimeException("Simulated
read failure"));
+
+ // Purgatory re-invokes tryComplete; get() throws, and the error
handling (exception handler)
+ // then throws too. The finally must still release the partition locks
before the throw propagates.
Review Comment:
Done.
--
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]
