chia7712 commented on code in PR #18254:
URL: https://github.com/apache/kafka/pull/18254#discussion_r1898879221
##########
core/src/test/java/kafka/clients/consumer/ConsumerIntegrationTest.java:
##########
@@ -185,4 +194,72 @@ public void
onPartitionsAssigned(Collection<TopicPartition> partitions) {
}
}
}
+
+ @ClusterTest(types = {Type.KRAFT}, brokers = 3)
+ public void testLeaderEpoch(ClusterInstance clusterInstance) throws
Exception {
+ String topic = "test-topic";
+ clusterInstance.createTopic(topic, 1, (short) 2);
+ var msgNum = 10;
+ sendMsg(clusterInstance, topic, msgNum);
+
+ try (var consumer = clusterInstance.consumer()) {
+ List<TopicPartition> topicPartitions =
Collections.singletonList(new TopicPartition(topic, 0));
Review Comment:
`List.of`
##########
core/src/test/java/kafka/clients/consumer/ConsumerIntegrationTest.java:
##########
@@ -185,4 +194,72 @@ public void
onPartitionsAssigned(Collection<TopicPartition> partitions) {
}
}
}
+
+ @ClusterTest(types = {Type.KRAFT}, brokers = 3)
+ public void testLeaderEpoch(ClusterInstance clusterInstance) throws
Exception {
+ String topic = "test-topic";
+ clusterInstance.createTopic(topic, 1, (short) 2);
+ var msgNum = 10;
+ sendMsg(clusterInstance, topic, msgNum);
+
+ try (var consumer = clusterInstance.consumer()) {
+ List<TopicPartition> topicPartitions =
Collections.singletonList(new TopicPartition(topic, 0));
+ consumer.assign(topicPartitions);
+ consumer.seekToBeginning(Collections.singletonList(new
TopicPartition(topic, 0)));
+
+ int consumed = 0;
+ while (consumed < msgNum) {
+ ConsumerRecords<Object, Object> records =
consumer.poll(Duration.ofMillis(1000));
+ for (ConsumerRecord<Object, Object> record : records) {
+ assertTrue(record.leaderEpoch().isPresent());
+ assertEquals(0, record.leaderEpoch().get());
+ }
+ consumed += records.count();
+ }
+
+ // make the leader epoch increment by shutdown the leader broker
+ shutdownFirstPartitionLeader(clusterInstance, topic);
+
+ sendMsg(clusterInstance, topic, msgNum);
+
+ consumed = 0;
+ while (consumed < msgNum) {
+ ConsumerRecords<Object, Object> records =
consumer.poll(Duration.ofMillis(1000));
+ for (ConsumerRecord<Object, Object> record : records) {
+ assertTrue(record.leaderEpoch().isPresent());
+ assertEquals(1, record.leaderEpoch().get());
+ }
+ consumed += records.count();
+ }
+ }
+ }
+
+ private void shutdownFirstPartitionLeader(ClusterInstance clusterInstance,
+ String topic) throws Exception {
+ var leaderBrokerId = -1;
+ try (var admin = clusterInstance.admin()) {
+ DescribeTopicsResult result = admin.describeTopics(List.of(topic));
Review Comment:
finding the leader is a common op, so could you please add a helper to
`ClusterInstance` to return leader of partition?
--
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
