apoorvmittal10 commented on code in PR #22881:
URL: https://github.com/apache/kafka/pull/22881#discussion_r3615087827
##########
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:
Yeah, thanks.
##########
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:
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]
