apoorvmittal10 commented on code in PR #22881:
URL: https://github.com/apache/kafka/pull/22881#discussion_r3615081999
##########
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:
Once when all partitions are acquired and local log partitions are released
in `maybeProcessRemoteFetch` and second when more bytes are needed to be
fetched then partitions are reacquired and relased in
`completeRemoteStorageShareFetchRequest -> readAndProcessFetchResultCompletion`
--
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]
