You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "brandboat (via GitHub)" <gi...@apache.org> on 2024/11/17 09:37:18 UTC
[PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
brandboat opened a new pull request, #17840:
URL: https://github.com/apache/kafka/pull/17840
related to https://issues.apache.org/jira/browse/KAFKA-18003, add test to make sure deleteRecords can handle the corrupted records. test steps are as follows:
1. create topic with 1 partition and 1 replica
1. produces 2 segment, each have 10 records
2. manually load the segment file to corrupt the data
3. failed to consume the all records due to corrupted record
4. add follower to topic
5. call `Admin.deleteRecords` to skip the corrupted record
6. check if the newly added follower is in-sync replica
7. succeed to consume the all records due to corrupted record
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003: add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya merged PR #17840:
URL: https://github.com/apache/kafka/pull/17840
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003: add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1986365438
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
Review Comment:
open https://issues.apache.org/jira/browse/KAFKA-18949 to trace 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845406011
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
+ val config = new Properties()
+ config.put("segment.ms", "1000")
Review Comment:
Thanks for the detailed review 😃 , I'll do this in the next commit.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845466023
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
Review Comment:
I realized that DeleteRecordsRequestTest focuses on testing `DeleteRecordsRequest`, so it might be better to place testDeleteRecordsAfterCorruptRecords to PlaintextAdminIntegrationTest, as it uses admin#deleteRecords without directly invoking DeleteRecordsRequest. WDYT, @divijvaidya?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845404340
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
+ val config = new Properties()
+ config.put("segment.ms", "1000")
+ createTopic(topic, numPartitions = 1, replicationFactor = 1, config)
+
+ client = createAdminClient
+
+ val consumer = createConsumer()
+ subscribeAndWaitForAssignment(topic, consumer)
+
+ val producer = createProducer()
Review Comment:
we need to close this and other closeables in a try-with-resources
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
+ val config = new Properties()
+ config.put("segment.ms", "1000")
Review Comment:
can we change the logic to rotate based on size? That will help us remove the Time.sleep in this test
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
Review Comment:
Should this test go in `DeleteRecordsRequestTest`?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
+ val config = new Properties()
+ config.put("segment.ms", "1000")
+ createTopic(topic, numPartitions = 1, replicationFactor = 1, config)
+
+ client = createAdminClient
+
+ val consumer = createConsumer()
+ subscribeAndWaitForAssignment(topic, consumer)
+
+ val producer = createProducer()
+ def sendRecords(begin: Int, end: Int) = {
+ val futures = (begin until end).map( i => {
+ val record = new ProducerRecord(topicPartition.topic, topicPartition.partition, s"$i".getBytes, s"$i".getBytes)
+ producer.send(record)
+ })
+ futures.foreach(_.get)
+ }
+ // create two segments, each have 10 records
+ sendRecords(0, 10)
+ Thread.sleep(1001)
+ sendRecords(10, 20)
+
+ val topicDesc = client.describeTopics(Collections.singletonList(topicPartition.topic()))
+ .allTopicNames().get().get(topicPartition.topic())
+ assertEquals(1, topicDesc.partitions().size())
+ val partitionLeaderId = topicDesc.partitions().get(0).leader().id()
+ val logDirMap = client.describeLogDirs(Collections.singletonList(partitionLeaderId))
+ .allDescriptions().get().get(partitionLeaderId)
+ val logDir = logDirMap.entrySet.stream
+ .filter(entry => entry.getValue.replicaInfos.containsKey(topicPartition)).findAny().get().getKey
+ // retrieve the path of the first segment
+ val logFilePath = LogFileUtils.logFile(Paths.get(logDir).resolve(topicPartition.toString).toFile, 0).toPath
+ val records = FileRecords.open(logFilePath.toFile)
+ assertEquals(10, records.records().asScala.iterator.size)
+
+ // manually load the inactive segment file to corrupt the data
+ val originalContent = Files.readAllBytes(logFilePath)
+ val newContent = ByteBuffer.allocate(JLong.BYTES + Integer.BYTES + originalContent.length)
+ newContent.putLong(0) // offset
+ newContent.putInt(0) // size -> this will make FileLogInputStream throw "Found record size 0 smaller than minimum record..."
+ newContent.put(Files.readAllBytes(logFilePath))
+ newContent.flip()
+ Files.write(logFilePath, newContent.array(), StandardOpenOption.TRUNCATE_EXISTING)
+
+ consumer.seekToBeginning(Collections.singletonList(topicPartition))
+ assertEquals("Encountered corrupt message when fetching offset 0 for topic-partition topic-0",
+ assertThrows(classOf[KafkaException], () => consumer.poll(JDuration.ofMillis(DEFAULT_MAX_WAIT_MS))).getMessage)
+
+ val partitionFollowerId = brokers.map(b => b.config.nodeId).filter(id => id != partitionLeaderId).head
+ val newAssignment = Map(topicPartition -> Optional.of(new NewPartitionReassignment(
+ List(Integer.valueOf(partitionLeaderId), Integer.valueOf(partitionFollowerId)).asJava))).asJava
+
+ // add follower to topic partition
+ client.alterPartitionReassignments(newAssignment).all().get()
+ // delete records in corrupt segment (the first segment)
+ client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(10L)).asJava).all.get
+ // verify reassignment is finished after delete records
+ TestUtils.waitUntilTrue(() => {
+ val isr = client.describeTopics(TopicCollection.ofTopicNames(List(topicPartition.topic()).asJava))
+ .allTopicNames().get().get(topic).partitions().get(0).isr().asScala.map(node => node.id())
+ isr.equals(List(partitionLeaderId, partitionFollowerId))
+ }, "The isr is incorrect")
Review Comment:
Consider using `TestUtils.waitForBrokersInIsr`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845353488
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
Review Comment:
I didn't use `getTestQuorumAndGroupProtocolParametersAll` here as we found that under async consumer, it wont throw exception in [L1598](https://github.com/apache/kafka/pull/17840/files#diff-41e66d9de25929a78b95f153b1ea93476851a379085d4eb7a6d118459f711b1aR1598)
```java
assertEquals("Encountered corrupt message when fetching offset 0 for topic-partition topic-0",
assertThrows(classOf[KafkaException], () => consumer.poll(JDuration.ofMillis(DEFAULT_MAX_WAIT_MS))).getMessage)
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1846165837
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
Review Comment:
I am a bit worried about `PlaintextAdminIntegrationTest.scala` becoming a very large class with varied different kinds of tests. Having said that, many other DeleteRecords tests reside here currently, so it is ok to add this one.
The refactoring of `PlaintextAdminIntegrationTest` can be left for some other time.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845407213
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
+ val config = new Properties()
+ config.put("segment.ms", "1000")
+ createTopic(topic, numPartitions = 1, replicationFactor = 1, config)
+
+ client = createAdminClient
+
+ val consumer = createConsumer()
+ subscribeAndWaitForAssignment(topic, consumer)
+
+ val producer = createProducer()
Review Comment:
Actually we already done that.
These two resources (consumer and producer) will be collected in
https://github.com/apache/kafka/blob/trunk/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala#L190
and
https://github.com/apache/kafka/blob/trunk/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala#L206
And will be closed in https://github.com/apache/kafka/blob/trunk/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala#L248
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18003 add test to make sure `Admin#deleteRecords` can handle the corrupted records [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #17840:
URL: https://github.com/apache/kafka/pull/17840#discussion_r1845407293
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1543,6 +1547,76 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNull(returnedOffsets.get(topicPartition))
}
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+ def testDeleteRecordsAfterCorruptRecords(quorum: String, groupProtocol: String): Unit = {
Review Comment:
Sounds good, thanks for the suggestion.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org