apoorvmittal10 commented on code in PR #17283:
URL: https://github.com/apache/kafka/pull/17283#discussion_r1793822101
##########
core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java:
##########
@@ -302,4 +295,54 @@ public void
testProcessFetchResponseWithLsoMovementForTopicPartition() {
Mockito.verify(sp1, times(0)).updateCacheAndOffsets(any(Long.class));
}
+ @Test
+ public void testProcessFetchResponseWhenNoRecordsAreAcquired() {
+ String groupId = "grp";
+ Uuid fooId = Uuid.randomUuid();
+
+ TopicIdPartition tp0 = new TopicIdPartition(fooId, new
TopicPartition("foo", 0));
+ Map<TopicIdPartition, Integer> partitionMaxBytes =
Collections.singletonMap(tp0, PARTITION_MAX_BYTES);
+
+ SharePartition sp0 = Mockito.mock(SharePartition.class);
+ Map<SharePartitionKey, SharePartition> partitionCacheMap =
Collections.singletonMap(new SharePartitionKey(groupId, tp0), sp0);
+
+ ShareFetchData shareFetchData = new ShareFetchData(
+ new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(),
FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
+ 1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK,
Optional.empty()),
+ groupId, Uuid.randomUuid().toString(), new CompletableFuture<>(),
partitionMaxBytes);
+
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ // Mock the replicaManager.fetchOffsetForTimestamp method to return a
timestamp and offset for the topic partition.
+ FileRecords.TimestampAndOffset timestampAndOffset = new
FileRecords.TimestampAndOffset(100L, 1L, Optional.empty());
+ doReturn(new OffsetResultHolder(Option.apply(timestampAndOffset),
Option.empty())).when(replicaManager).fetchOffsetForTimestamp(any(TopicPartition.class),
anyLong(), any(), any(), anyBoolean());
+
+ when(sp0.nextFetchOffset()).thenReturn((long) 0);
+ when(sp0.acquire(anyString(),
any(FetchPartitionData.class))).thenReturn(Collections.emptyList());
+ doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));
+
+ MemoryRecords records = MemoryRecords.withRecords(Compression.NONE,
+ new SimpleRecord("0".getBytes(), "v".getBytes()),
+ new SimpleRecord("1".getBytes(), "v".getBytes()),
+ new SimpleRecord("2".getBytes(), "v".getBytes()),
+ new SimpleRecord(null, "value".getBytes()));
+
+ Map<TopicIdPartition, FetchPartitionData> responseData =
Collections.singletonMap(
+ tp0, new FetchPartitionData(Errors.OFFSET_OUT_OF_RANGE, 0L, 0L,
Review Comment:
My bad, I have added other scenrio with empty as well.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
