clolov commented on code in PR #14049:
URL: https://github.com/apache/kafka/pull/14049#discussion_r1276037068
##########
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java:
##########
@@ -168,4 +169,64 @@ private void waitUntilConsumerCatchesup(TopicIdPartition
newLeaderTopicIdPartiti
}
}
+ @Test
+ public void testRemoteLogSizeCalculationForUnknownTopicIdPartitionThrows()
{
+ TopicIdPartition topicIdPartition = new
TopicIdPartition(Uuid.randomUuid(), new TopicPartition("singleton", 0));
+ Assertions.assertThrows(RemoteResourceNotFoundException.class, () ->
topicBasedRlmm().remoteLogSize(topicIdPartition, 0));
+ }
+
+ @Test
+ public void testRemoteLogSizeCalculationWithSegmentsOfTheSameEpoch()
throws RemoteStorageException, TimeoutException {
+ TopicIdPartition topicIdPartition = new
TopicIdPartition(Uuid.randomUuid(), new TopicPartition("singleton", 0));
+ TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager
= topicBasedRlmm();
+
+ RemoteLogSegmentMetadata firstSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 0, 100, -1L, 0, time.milliseconds(), SEG_SIZE,
Collections.singletonMap(0, 0L));
+ RemoteLogSegmentMetadata secondSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 100, 200, -1L, 0, time.milliseconds(), SEG_SIZE * 2,
Collections.singletonMap(0, 0L));
+ RemoteLogSegmentMetadata thirdSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 200, 300, -1L, 0, time.milliseconds(), SEG_SIZE * 3,
Collections.singletonMap(0, 0L));
+
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(firstSegmentMetadata);
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(secondSegmentMetadata);
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(thirdSegmentMetadata);
+
+
topicBasedRemoteLogMetadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition),
Collections.emptySet());
+
+ // RemoteLogSegmentMetadata events are already published, and
topicBasedRlmm's consumer manager will start
+ // fetching those events and build the cache.
+ waitUntilConsumerCatchesup(topicIdPartition, topicIdPartition,
30_000L);
+
+ Long remoteLogSize =
topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 0);
+
+ Assertions.assertEquals(SEG_SIZE * 6, remoteLogSize);
+ }
+
+ @Test
+ public void testRemoteLogSizeCalculationWithSegmentsOfDifferentEpochs()
throws RemoteStorageException, TimeoutException {
+ TopicIdPartition topicIdPartition = new
TopicIdPartition(Uuid.randomUuid(), new TopicPartition("singleton", 0));
+ TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager
= topicBasedRlmm();
+
+ RemoteLogSegmentMetadata firstSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 0, 100, -1L, 0, time.milliseconds(), SEG_SIZE,
Collections.singletonMap(0, 0L));
+ RemoteLogSegmentMetadata secondSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 100, 200, -1L, 0, time.milliseconds(), SEG_SIZE * 2,
Collections.singletonMap(1, 100L));
+ RemoteLogSegmentMetadata thirdSegmentMetadata = new
RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition,
Uuid.randomUuid()),
+ 200, 300, -1L, 0, time.milliseconds(), SEG_SIZE * 3,
Collections.singletonMap(2, 200L));
+
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(firstSegmentMetadata);
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(secondSegmentMetadata);
+
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(thirdSegmentMetadata);
+
+
topicBasedRemoteLogMetadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition),
Collections.emptySet());
+
+ // RemoteLogSegmentMetadata events are already published, and
topicBasedRlmm's consumer manager will start
+ // fetching those events and build the cache.
+ waitUntilConsumerCatchesup(topicIdPartition, topicIdPartition,
30_000L);
+
+ Assertions.assertEquals(SEG_SIZE,
topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 0));
+ Assertions.assertEquals(SEG_SIZE * 2,
topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 1));
+ Assertions.assertEquals(SEG_SIZE * 3,
topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 2));
Review Comment:
Hopefully the test below addresses this. Let me know if not and I will alter
it!
--
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
