showuon commented on code in PR #15616:
URL: https://github.com/apache/kafka/pull/15616#discussion_r1575591635
##########
core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala:
##########
@@ -116,6 +124,56 @@ class AlterReplicaLogDirsRequestTest extends
BaseRequestTest {
assertEquals(Errors.KAFKA_STORAGE_ERROR,
findErrorForPartition(alterReplicaDirResponse3, new TopicPartition(topic, 2)))
}
+ @Test
+ def testAlterReplicaLogDirsRequestWithRetention(): Unit = {
+ val partitionNum = 1
+
+ // Alter replica dir before topic creation
+ val logDir1 = new File(servers.head.config.logDirs(1)).getAbsolutePath
+ val partitionDirs1 = (0 until partitionNum).map(partition => new
TopicPartition(topic, partition) -> logDir1).toMap
+ val alterReplicaLogDirsResponse1 =
sendAlterReplicaLogDirsRequest(partitionDirs1)
+
+ // The response should show error UNKNOWN_TOPIC_OR_PARTITION for all
partitions
+ val tp = new TopicPartition(topic, 0)
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION,
findErrorForPartition(alterReplicaLogDirsResponse1, tp))
+ assertTrue(servers.head.logManager.getLog(tp).isEmpty)
+
+ val topicProperties = new Properties()
+ topicProperties.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024")
+ topicProperties.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, "10000")
+ topicProperties.put(TopicConfig.SEGMENT_BYTES_CONFIG, "1024")
+
+ createTopic(topic, partitionNum, 1, topicProperties)
+ assertEquals(logDir1, servers.head.logManager.getLog(tp).get.dir.getParent)
+
+ // send enough records to trigger log rolling
+ (0 until 20).foreach { _ =>
+ TestUtils.generateAndProduceMessages(servers, topic, 10, 1)
+ }
+ TestUtils.waitUntilTrue(() => servers.head.logManager.getLog(new
TopicPartition(topic, 0)).get.numberOfSegments > 1,
+ "timed out waiting for log segment to roll")
+
+ // Wait for log segment retention. Override initialTaskDelayMs as 5
seconds.
+ // The first retention task is executed after 5 seconds, so waiting for 10
seconds should be enough.
Review Comment:
This comment is not correct now.
##########
core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala:
##########
@@ -116,6 +124,56 @@ class AlterReplicaLogDirsRequestTest extends
BaseRequestTest {
assertEquals(Errors.KAFKA_STORAGE_ERROR,
findErrorForPartition(alterReplicaDirResponse3, new TopicPartition(topic, 2)))
}
+ @Test
+ def testAlterReplicaLogDirsRequestWithRetention(): Unit = {
+ val partitionNum = 1
+
+ // Alter replica dir before topic creation
+ val logDir1 = new File(servers.head.config.logDirs(1)).getAbsolutePath
+ val partitionDirs1 = (0 until partitionNum).map(partition => new
TopicPartition(topic, partition) -> logDir1).toMap
+ val alterReplicaLogDirsResponse1 =
sendAlterReplicaLogDirsRequest(partitionDirs1)
+
+ // The response should show error UNKNOWN_TOPIC_OR_PARTITION for all
partitions
+ val tp = new TopicPartition(topic, 0)
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION,
findErrorForPartition(alterReplicaLogDirsResponse1, tp))
+ assertTrue(servers.head.logManager.getLog(tp).isEmpty)
+
+ val topicProperties = new Properties()
+ topicProperties.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024")
+ topicProperties.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, "10000")
Review Comment:
I see. Make sense. It's for L158's assertion, right? Could you add a comment
here to explain 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
