adixitconfluent commented on code in PR #22881:
URL: https://github.com/apache/kafka/pull/22881#discussion_r3614689990
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -1488,6 +1547,88 @@ public void
testRemoteStorageFetchTryCompleteReturnsFalse() {
delayedShareFetch.lock().unlock();
}
+ /**
+ * Validates the tryComplete() that handles an in-flight remote fetch:
when tryComplete() is
+ * re-invoked (e.g. via a purgatory checkAndComplete) while a remote fetch
initiated by an earlier
+ * tryComplete() is still pending, it must route to
maybeCompletePendingRemoteFetch() rather than
+ * re-acquiring partitions or starting a new read. It stays in purgatory
while the remote fetch is
+ * pending and completes once the remote fetch is done.
+ */
+ @SuppressWarnings("unchecked")
Review Comment:
nit: I think this is unnecessary
##########
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:
nit: "the finally block" instead of "the finally"
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -2020,6 +2161,216 @@ public void
testRemoteStorageFetchRequestCompletionAlongWithLocalLogRead() {
}
}
+ /**
+ * Tests that when a remote fetch completes alongside a successful
additional local read (with data),
+ * the remote-fetch completion path does not re-enqueue the local
partitions to the action queue.
+ * The local partitions' completion is separately triggered by
processFetchResultAndComplete, so
+ * completeRemoteFetchRequest must only enqueue the remote (acquired)
partitions.
+ */
+ @Test
+ public void
testRemoteStorageFetchCompletionDoesNotDuplicateLocalPartitionsInActionQueue() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 1));
+ TopicIdPartition tp2 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 2));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(true);
+ when(sp2.canAcquireRecords()).thenReturn(true);
+
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new
LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
+
+ CompletableFuture<Map<TopicIdPartition,
ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, "grp",
Uuid.randomUuid().toString(),
+ future, List.of(tp0, tp1, tp2), BATCH_OPTIMIZED, BATCH_SIZE,
MAX_FETCH_RECORDS,
+ BROKER_TOPIC_STATS);
+
+ when(sp0.nextFetchOffset()).thenReturn(10L);
+ when(sp1.nextFetchOffset()).thenReturn(20L);
+ when(sp2.nextFetchOffset()).thenReturn(30L);
+ when(sp0.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp1.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp2.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+
+ try (MockedStatic<ShareFetchUtils> mockedShareFetchUtils =
Mockito.mockStatic(ShareFetchUtils.class)) {
+ Map<TopicIdPartition, ShareFetchResponseData.PartitionData>
partitionDataMap = new LinkedHashMap<>();
+ partitionDataMap.put(tp0,
mock(ShareFetchResponseData.PartitionData.class));
+ partitionDataMap.put(tp1,
mock(ShareFetchResponseData.PartitionData.class));
+ partitionDataMap.put(tp2,
mock(ShareFetchResponseData.PartitionData.class));
+ mockedShareFetchUtils.when(() ->
ShareFetchUtils.processFetchResponse(any(), any(), any(), any(),
any())).thenReturn(partitionDataMap);
+
+ // First read (tryComplete): tp0, tp1 local and tp2 remote. Second
read (additional local read
+ // in onComplete): tp0, tp1 returned WITH data so that they appear
in partitionsWithData.
+ doAnswer(invocation -> buildLocalAndRemoteFetchResult(Set.of(tp0,
tp1), Set.of(tp2))
+ ).doAnswer(invocation -> buildLogReadResult(List.of(tp0, tp1))
+ ).when(replicaManager).readFromLog(any(), any(),
any(ReplicaQuota.class), anyBoolean());
+
+ // Remote fetch object completes within tryComplete so the request
moves on to onComplete.
+ RemoteLogReadResult remoteFetchResult = new
RemoteLogReadResult(Optional.of(REMOTE_FETCH_INFO), Optional.empty());
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+ doAnswer(invocationOnMock -> {
+ Consumer<RemoteLogReadResult> callback =
invocationOnMock.getArgument(1);
+ callback.accept(remoteFetchResult);
+ return CompletableFuture.completedFuture(remoteFetchResult);
+ }).when(remoteLogManager).asyncRead(any(), any());
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ Partition p0 = mock(Partition.class);
+ when(p0.isLeader()).thenReturn(true);
+ Partition p1 = mock(Partition.class);
+ when(p1.isLeader()).thenReturn(true);
+ Partition p2 = mock(Partition.class);
+ when(p2.isLeader()).thenReturn(true);
+
when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
+
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);
+
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);
+
+ // Execute the action queue runnables synchronously so
completeDelayedShareFetchRequest is invoked.
+ doAnswer(invocation -> {
+ Runnable runnable = invocation.getArgument(0);
+ runnable.run();
+ return null;
+ }).when(replicaManager).addToActionQueue(any());
+
+ Uuid fetchId = Uuid.randomUuid();
+ DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withReplicaManagerLogReader(replicaManager)
+ .withSharePartitions(sharePartitions)
+
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0, tp1, tp2)))
+ .withFetchId(fetchId)
+ .build());
+
+ when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp2.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+ assertTrue(shareFetch.isCompleted());
+
+ // The local partitions tp0 and tp1 are enqueued twice: once when
they are released as
+ // non-remote partitions while the remote fetch is prepared in
tryComplete, and again
+ // after they are re-acquired for the additional local read in
onComplete (via
+ // processFetchResultAndComplete). The remote-fetch completion
(completeRemoteFetchRequest)
+ // must not enqueue them a third time - it only enqueues the
remote (acquired) partitions.
+ Mockito.verify(replicaManager,
times(2)).completeDelayedShareFetchRequest(
Review Comment:
I think we should add the asserts for number of calls to
maybeAcquireFetchLock and releaseFetchLock for tp0, tp1 and tp2 here
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -2020,6 +2161,216 @@ public void
testRemoteStorageFetchRequestCompletionAlongWithLocalLogRead() {
}
}
+ /**
+ * Tests that when a remote fetch completes alongside a successful
additional local read (with data),
+ * the remote-fetch completion path does not re-enqueue the local
partitions to the action queue.
+ * The local partitions' completion is separately triggered by
processFetchResultAndComplete, so
+ * completeRemoteFetchRequest must only enqueue the remote (acquired)
partitions.
+ */
+ @Test
+ public void
testRemoteStorageFetchCompletionDoesNotDuplicateLocalPartitionsInActionQueue() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 1));
+ TopicIdPartition tp2 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 2));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(true);
+ when(sp2.canAcquireRecords()).thenReturn(true);
+
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new
LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
+
+ CompletableFuture<Map<TopicIdPartition,
ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, "grp",
Uuid.randomUuid().toString(),
+ future, List.of(tp0, tp1, tp2), BATCH_OPTIMIZED, BATCH_SIZE,
MAX_FETCH_RECORDS,
+ BROKER_TOPIC_STATS);
+
+ when(sp0.nextFetchOffset()).thenReturn(10L);
+ when(sp1.nextFetchOffset()).thenReturn(20L);
+ when(sp2.nextFetchOffset()).thenReturn(30L);
+ when(sp0.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp1.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp2.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+
+ try (MockedStatic<ShareFetchUtils> mockedShareFetchUtils =
Mockito.mockStatic(ShareFetchUtils.class)) {
+ Map<TopicIdPartition, ShareFetchResponseData.PartitionData>
partitionDataMap = new LinkedHashMap<>();
+ partitionDataMap.put(tp0,
mock(ShareFetchResponseData.PartitionData.class));
+ partitionDataMap.put(tp1,
mock(ShareFetchResponseData.PartitionData.class));
+ partitionDataMap.put(tp2,
mock(ShareFetchResponseData.PartitionData.class));
+ mockedShareFetchUtils.when(() ->
ShareFetchUtils.processFetchResponse(any(), any(), any(), any(),
any())).thenReturn(partitionDataMap);
+
+ // First read (tryComplete): tp0, tp1 local and tp2 remote. Second
read (additional local read
+ // in onComplete): tp0, tp1 returned WITH data so that they appear
in partitionsWithData.
+ doAnswer(invocation -> buildLocalAndRemoteFetchResult(Set.of(tp0,
tp1), Set.of(tp2))
+ ).doAnswer(invocation -> buildLogReadResult(List.of(tp0, tp1))
+ ).when(replicaManager).readFromLog(any(), any(),
any(ReplicaQuota.class), anyBoolean());
+
+ // Remote fetch object completes within tryComplete so the request
moves on to onComplete.
+ RemoteLogReadResult remoteFetchResult = new
RemoteLogReadResult(Optional.of(REMOTE_FETCH_INFO), Optional.empty());
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+ doAnswer(invocationOnMock -> {
+ Consumer<RemoteLogReadResult> callback =
invocationOnMock.getArgument(1);
+ callback.accept(remoteFetchResult);
+ return CompletableFuture.completedFuture(remoteFetchResult);
+ }).when(remoteLogManager).asyncRead(any(), any());
+
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
+
+ Partition p0 = mock(Partition.class);
+ when(p0.isLeader()).thenReturn(true);
+ Partition p1 = mock(Partition.class);
+ when(p1.isLeader()).thenReturn(true);
+ Partition p2 = mock(Partition.class);
+ when(p2.isLeader()).thenReturn(true);
+
when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
+
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);
+
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);
+
+ // Execute the action queue runnables synchronously so
completeDelayedShareFetchRequest is invoked.
+ doAnswer(invocation -> {
+ Runnable runnable = invocation.getArgument(0);
+ runnable.run();
+ return null;
+ }).when(replicaManager).addToActionQueue(any());
+
+ Uuid fetchId = Uuid.randomUuid();
+ DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withReplicaManagerLogReader(replicaManager)
+ .withSharePartitions(sharePartitions)
+
.withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0, tp1, tp2)))
+ .withFetchId(fetchId)
+ .build());
+
+ when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp2.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+ assertTrue(shareFetch.isCompleted());
+
+ // The local partitions tp0 and tp1 are enqueued twice: once when
they are released as
Review Comment:
For the first scenario, can you please mention the function trail by which
completeDelayedShareFetchRequest is getting called
##########
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:
nit: "the finally block" instead of "the finally"
--
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]
