apoorvmittal10 commented on code in PR #17283:
URL: https://github.com/apache/kafka/pull/17283#discussion_r1795215724
##########
core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java:
##########
@@ -300,4 +293,69 @@ 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);
+ SharePartitionManager sharePartitionManager =
mock(SharePartitionManager.class);
+ when(sharePartitionManager.sharePartition(groupId,
tp0)).thenReturn(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.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()));
+
+ // When no records are acquired from share partition.
+ Map<TopicIdPartition, FetchPartitionData> responseData =
Collections.singletonMap(
+ tp0, new FetchPartitionData(Errors.NONE, 0L, 0L,
+ records, Optional.empty(), OptionalLong.empty(),
Optional.empty(),
+ OptionalInt.empty(), false));
+
+ Map<TopicIdPartition, ShareFetchResponseData.PartitionData> resultData
=
+ ShareFetchUtils.processFetchResponse(shareFetchData, responseData,
sharePartitionManager, replicaManager);
+
+ assertEquals(1, resultData.size());
+ assertTrue(resultData.containsKey(tp0));
+ assertEquals(0, resultData.get(tp0).partitionIndex());
+ assertNull(resultData.get(tp0).records());
+ assertTrue(resultData.get(tp0).acquiredRecords().isEmpty());
+ assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
+
+ // When fetch partition data has OFFSET_OUT_OF_RANGE error.
+ responseData = Collections.singletonMap(
+ tp0, new FetchPartitionData(Errors.OFFSET_OUT_OF_RANGE, 0L, 0L,
+ records, Optional.empty(), OptionalLong.empty(),
Optional.empty(),
+ OptionalInt.empty(), false));
+
+ resultData = ShareFetchUtils.processFetchResponse(shareFetchData,
responseData, sharePartitionManager, replicaManager);
+
+ assertEquals(1, resultData.size());
+ assertTrue(resultData.containsKey(tp0));
+ assertEquals(0, resultData.get(tp0).partitionIndex());
+ assertNull(resultData.get(tp0).records());
+ assertTrue(resultData.get(tp0).acquiredRecords().isEmpty());
+ assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
+
+ Mockito.verify(sp0, times(1)).updateCacheAndOffsets(any(Long.class));
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
