adixitconfluent commented on code in PR #17539:
URL: https://github.com/apache/kafka/pull/17539#discussion_r1828775451
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -107,23 +109,125 @@ public void
testDelayedShareFetchTryCompleteReturnsFalse() {
when(sp0.maybeAcquireFetchLock()).thenReturn(true);
when(sp1.maybeAcquireFetchLock()).thenReturn(true);
- SharePartitionManager sharePartitionManager =
mock(SharePartitionManager.class);
- when(sharePartitionManager.sharePartition(groupId,
tp0)).thenReturn(sp0);
- when(sharePartitionManager.sharePartition(groupId,
tp1)).thenReturn(sp1);
+ Map<TopicIdPartition, SharePartition> sharePartitions = new
HashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
ShareFetchData shareFetchData = new ShareFetchData(FETCH_PARAMS,
groupId, Uuid.randomUuid().toString(),
new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.canAcquireRecords()).thenReturn(false);
when(sp1.canAcquireRecords()).thenReturn(false);
- DelayedShareFetch delayedShareFetch =
DelayedShareFetchBuilder.builder()
+ DelayedShareFetch delayedShareFetch =
spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetchData)
- .withSharePartitionManager(sharePartitionManager)
- .build();
+ .withSharePartitions(sharePartitions)
+ .build());
// Since there is no partition that can be acquired, tryComplete
should return false.
assertFalse(delayedShareFetch.tryComplete());
assertFalse(delayedShareFetch.isCompleted());
+ Mockito.verify(delayedShareFetch,
times(0)).releasePartitionLocks(any());
+ }
+
+ @Test
+ public void
testDelayedShareFetchTryCompleteReturnsFalseDueToMinBytesOnFirstSharePartitionFetch()
{
+ String groupId = "grp";
+ Uuid topicId = Uuid.randomUuid();
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(topicId, new
TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(topicId, new
TopicPartition("foo", 1));
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+
+ when(sp0.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock()).thenReturn(true);
+
+ Map<TopicIdPartition, SharePartition> sharePartitions = new
HashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+
+ ShareFetchData shareFetchData = new ShareFetchData(
+ new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(),
FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
+ 2, 1024 * 1024, FetchIsolation.HIGH_WATERMARK,
Optional.empty()), groupId, Uuid.randomUuid().toString(),
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
+
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(false);
+ when(sp0.acquire(any(), anyInt(), any())).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new
ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short)
1)));
+
+ // We are testing the case when the share partition is getting fetched
for the first time, hence we are using 1
+ // as the file position, so it doesn't satisfy the minBytes(2).
+ when(sp0.fetchOffsetMetadata()).thenReturn(Optional.of(new
LogOffsetMetadata(0, 1, 0)));
Review Comment:
my bad, you're right, I've adjusted the mock to return `Optional.empty()`
for the first time and `Optional.of(new LogOffsetMetadata(0, 1, 0))` for the
second time (post `readFromLog` call)
--
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]
