fvaleri commented on code in PR #16475:
URL: https://github.com/apache/kafka/pull/16475#discussion_r1666515845
##########
core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala:
##########
@@ -243,6 +246,209 @@ class DumpLogSegmentsTest {
assertEquals(Map.empty, errors.shallowOffsetNotFound)
}
+ @Test
+ def testDumpRemoteLogMetadataEmpty(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024)
+ log = LogTestUtils.createLog(logDir, logConfig, new BrokerTopicStats,
time.scheduler, time)
+
+ val output = runDumpLogSegments(Array("--remote-log-metadata-decoder",
"--files", logFilePath))
+ assertTrue(output.contains("Log starting offset: 0"))
+ }
+
+ @Test
+ def testDumpRemoteLogMetadataOneRecordOneBatch(): Unit = {
+ val topicId = Uuid.randomUuid
+ val topicName = "foo"
+
+ val metadata = Seq(new RemotePartitionDeleteMetadata(new
TopicIdPartition(topicId, new TopicPartition(topicName, 0)),
+ RemotePartitionDeleteState.DELETE_PARTITION_MARKED, time.milliseconds,
0))
+
+ val records: Array[SimpleRecord] = metadata.map(message => {
+ new SimpleRecord(null, new RemoteLogMetadataSerde().serialize(message))
+ }).toArray
+
+ val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024)
+ log = LogTestUtils.createLog(logDir, logConfig, new BrokerTopicStats,
time.scheduler, time)
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
records:_*), leaderEpoch = 0)
+ log.flush(false)
+
+ val expectedDeletePayload =
String.format("RemotePartitionDeleteMetadata{topicPartition=%s:%s-0, " +
+ "state=DELETE_PARTITION_MARKED, eventTimestampMs=0, brokerId=0}",
topicId, topicName)
+
+ val output = runDumpLogSegments(Array("--remote-log-metadata-decoder",
"--files", logFilePath))
+ assertTrue(output.contains("Log starting offset: 0"))
+ assertTrue(output.contains(expectedDeletePayload))
+ }
+
+ @Test
+ def testDumpRemoteLogMetadataMultipleRecordsOneBatch(): Unit = {
+ val topicId = Uuid.randomUuid
+ val topicName = "foo"
+ val remoteSegmentId = Uuid.randomUuid
+
+ val topicIdPartition = new TopicIdPartition(topicId, new
TopicPartition(topicName, 0))
+ val remoteLogSegmentId = new RemoteLogSegmentId(topicIdPartition,
remoteSegmentId)
+
+ val metadata = Seq(new RemoteLogSegmentMetadataUpdate(remoteLogSegmentId,
time.milliseconds,
+ Optional.of(new RemoteLogSegmentMetadata.CustomMetadata(Array[Byte](0,
1, 2, 3))), RemoteLogSegmentState.COPY_SEGMENT_FINISHED, 0),
+ new RemotePartitionDeleteMetadata(topicIdPartition,
RemotePartitionDeleteState.DELETE_PARTITION_MARKED, time.milliseconds, 0))
+
+ val metadataRecords: Array[SimpleRecord] = metadata.map(message => {
+ new SimpleRecord(null, new RemoteLogMetadataSerde().serialize(message))
+ }).toArray
+
+ val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024)
+ log = LogTestUtils.createLog(logDir, logConfig, new BrokerTopicStats,
time.scheduler, time)
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
metadataRecords:_*), leaderEpoch = 0)
+ log.flush(false)
+
+ val expectedUpdatePayload =
String.format("RemoteLogSegmentMetadataUpdate{remoteLogSegmentId=" +
+ "RemoteLogSegmentId{topicIdPartition=%s:%s-0, id=%s},
customMetadata=Optional[" +
+ "CustomMetadata{4 bytes}], state=COPY_SEGMENT_FINISHED,
eventTimestampMs=0, brokerId=0}", topicId, topicName, remoteSegmentId)
+ val expectedDeletePayload =
String.format("RemotePartitionDeleteMetadata{topicPartition=%s:%s-0, " +
+ "state=DELETE_PARTITION_MARKED, eventTimestampMs=0, brokerId=0}",
topicId, topicName)
+
+ val output = runDumpLogSegments(Array("--remote-log-metadata-decoder",
"--files", logFilePath))
+ assertTrue(output.contains("Log starting offset: 0"))
+ assertTrue(output.contains(expectedUpdatePayload))
+ assertTrue(output.contains(expectedDeletePayload))
+ }
+
+ @Test
+ def testDumpRemoteLogMetadataMultipleRecordsMultipleBatches(): Unit = {
+ val topicId = Uuid.randomUuid
+ val topicName = "foo"
+ val remoteSegmentId = Uuid.randomUuid
+
+ val topicIdPartition = new TopicIdPartition(topicId, new
TopicPartition(topicName, 0))
+ val remoteLogSegmentId = new RemoteLogSegmentId(topicIdPartition,
remoteSegmentId)
+
+ val metadata = Seq(
+ new RemoteLogSegmentMetadataUpdate(remoteLogSegmentId, time.milliseconds,
+ Optional.of(new RemoteLogSegmentMetadata.CustomMetadata(Array[Byte](0,
1, 2, 3))), RemoteLogSegmentState.COPY_SEGMENT_FINISHED, 0),
+ new RemotePartitionDeleteMetadata(topicIdPartition,
RemotePartitionDeleteState.DELETE_PARTITION_MARKED, time.milliseconds, 0)
+ )
+
+ val records: Array[SimpleRecord] = metadata.map(message => {
+ new SimpleRecord(null, new RemoteLogMetadataSerde().serialize(message))
+ }).toArray
+
+ val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024)
+ log = LogTestUtils.createLog(logDir, logConfig, new BrokerTopicStats,
time.scheduler, time)
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
records:_*), leaderEpoch = 0)
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
records:_*), leaderEpoch = 0)
+ log.flush(false)
+
+ val expectedUpdatePayload =
String.format("RemoteLogSegmentMetadataUpdate{remoteLogSegmentId=" +
+ "RemoteLogSegmentId{topicIdPartition=%s:%s-0, id=%s},
customMetadata=Optional[" +
+ "CustomMetadata{4 bytes}], state=COPY_SEGMENT_FINISHED,
eventTimestampMs=0, brokerId=0}", topicId, topicName, remoteSegmentId)
+ val expectedDeletePayload =
String.format("RemotePartitionDeleteMetadata{topicPartition=%s:%s-0, " +
+ "state=DELETE_PARTITION_MARKED, eventTimestampMs=0, brokerId=0}",
topicId, topicName)
+
+ val output = runDumpLogSegments(Array("--remote-log-metadata-decoder",
"--files", logFilePath))
+ assertTrue(output.contains("Log starting offset: 0"))
+ assertTrue(countSubstring(output, expectedUpdatePayload) == 2)
+ assertTrue(countSubstring(output, expectedDeletePayload) == 2)
Review Comment:
Yes, this makes sense. Thanks.
--
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]
