You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2026/03/05 14:19:27 UTC
[PR] KAFKA-20254: Add integration test for streams group migration with broker restart [kafka]
lucasbru opened a new pull request, #21642:
URL: https://github.com/apache/kafka/pull/21642
Depends on #21641.
This adds an integration test that reproduces the scenario where log compaction removes the GroupMetadata tombstone from __consumer_offsets, leaving orphan offset commit records before the streams group records. On broker restart, replaying these records creates a simple classic group that the streams group records must handle.
The test commits an offset for an "orphan" topic using the classic group's ID, then migrates to the streams protocol. After triggering log compaction to remove the tombstone, it restarts the broker and verifies the streams group still works. This exercises the isSimpleGroup() fix in getOrMaybeCreatePersistedStreamsGroup.
Also adds shutdownBroker/startBroker methods to EmbeddedKafkaCluster to enable broker restart within tests.
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2907787196
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
Good point, removed the lambda entirely — the wait-for-RUNNING logic is now inlined in `withStreamsApp`.
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
Agreed, inlined the wait-for-RUNNING logic into `withStreamsApp` and removed the lambda parameter.
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2897935030
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
Review Comment:
The newly added `rollAndCompactConsumerOffsets` doesn't do the trick?
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2895058356
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
Review Comment:
Removed.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
Review Comment:
Good idea. I've added a `leaveGroup` parameter to `processExactlyOneRecord` and use `CloseOptions.groupMembershipOperation(LEAVE_GROUP)` for the classic protocol step, which removes the need for `waitForEmptyConsumerGroup`.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
Review Comment:
It does commit for the input topic — but the orphan offset is for a different topic-partition that the streams app never subscribes to, so its offset commit record has a different key in `__consumer_offsets` and won't be overwritten. I've clarified the comment.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
Review Comment:
Agreed, I couldn't find a reliable way to assert compaction has run from the client side. I've added a comment noting the false-positive risk. As you said, worst case is a missing test signal, not a flaky failure.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
+
+ // Step 6: Restart the broker to force replay of __consumer_offsets.
+ CLUSTER.shutdownBroker(0);
+ CLUSTER.startBroker(0);
+
+ // Step 7: Verify the broker can still serve the streams group after replay.
+ processExactlyOneRecord(streamsBuilder, props, "3", "C");
+ }
+
+ private void commitOrphanOffset(final String groupId, final String topic) {
+ final Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+ try (final org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer =
Review Comment:
Fixed.
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2897994322
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,79 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedStreamsGroup(): Unit = {
Review Comment:
Are we sure the group was converted to "streams". -- In the Java test, we ensure that the app processes a record -- but here, we call `start()` and close quickly (not sure what `f(streams)` does...?)
In `testCoordinatorFailoverAfterCompactingPartitionWithUpgradedConsumerGroup` I only see topic `foo` but not orphaned topic, and `rollAndCompactConsumerOffsets()` is only called ones. Did you modify the test after you remove the fix for "consumer" case?
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
What is this line?
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #21642:
URL: https://github.com/apache/kafka/pull/21642#issuecomment-4031050899
@dajac thanks for the comments! Ready for re-review.
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #21642:
URL: https://github.com/apache/kafka/pull/21642
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2905567439
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
Review Comment:
Yes, switched to using it. I had to fix two issues to make the tests actually fail without the fix: (1) awaitCleaned(tp, 0, timeout) was returning immediately on the second call because the checkpoint was already past offset 0 — now uses logEndOffset as the threshold. (2) Tombstones need two compaction passes (first sets deleteHorizonMs, second removes the tombstone), but the second pass needs dirty data — now flooding between the two passes. Both tests now properly fail without the fix.
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2907415971
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
But not even sure, why we would need to pass in a lambda to begin with? Especially as it's the same for both test cases? What do we gain?
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2892525374
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
Review Comment:
Yes, that's what the next two lines of the code say :) -- Comments seems unncessary?
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2892555546
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
Review Comment:
Can we verify this differently. Using sleep looks questionable (I guess, it could be fine for this test, as if we don't sleep long enough, and compaction did not trigger yet, we would only get a false-positive result -- so we are not subject to a flaky test, but only a missing test signal, ie, we might pass even if there is a bug...)
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
+ final String orphanTopic = "orphan-" + safeTestName;
+ CLUSTER.createTopic(orphanTopic);
+ commitOrphanOffset(appId, orphanTopic);
+
+ // Step 3: Migrate to the streams protocol and process a record. This
+ // writes a GroupMetadata tombstone for the classic group followed by
+ // streams group records.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ processExactlyOneRecord(streamsBuilder, props, "2", "B");
+
+ // Step 4: Configure aggressive compaction so that the GroupMetadata
+ // tombstone is removed, leaving only the orphan offset commit record
+ // before the streams group records.
+ configureAggressiveCompaction();
+
+ // Step 5: Flood __consumer_offsets with offset commits to trigger segment
+ // rotation and compaction.
+ floodConsumerOffsetsForCompaction();
+
+ // Wait for compaction to clean up the GroupMetadata tombstone.
+ Thread.sleep(5000);
+
+ // Step 6: Restart the broker to force replay of __consumer_offsets.
+ CLUSTER.shutdownBroker(0);
+ CLUSTER.startBroker(0);
+
+ // Step 7: Verify the broker can still serve the streams group after replay.
+ processExactlyOneRecord(streamsBuilder, props, "3", "C");
+ }
+
+ private void commitOrphanOffset(final String groupId, final String topic) {
+ final Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+ try (final org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer =
Review Comment:
We should use an import...
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
Review Comment:
Why does the group not commit? We call `KafkaStreams#close()` inside `processExactlyOneRecord(...)` so a commit should have been triggered?
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2910470544
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -331,12 +512,34 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def writeOneOffsetCommit(): Unit = {
+ // Write a single offset commit to create dirty data past the cleaner
+ // checkpoint so the cleaner will re-process previously compacted
+ // segments on the next compaction pass.
+ withConsumer(groupId = "compaction-trigger", groupProtocol = GroupProtocol.CLASSIC, enableAutoCommit = false) { consumer =>
+ val tp = new TopicPartition("foo", 0)
+ consumer.assign(java.util.List.of(tp))
+ consumer.commitSync(java.util.Map.of(tp, new OffsetAndMetadata(0)))
+ }
+ }
+
+ private def configureDeleteRetention(): Unit = {
+ withAdmin { admin =>
+ val resource = new ConfigResource(ConfigResource.Type.TOPIC, "__consumer_offsets")
+ admin.incrementalAlterConfigs(java.util.Map.of(resource, java.util.List.of(
+ new AlterConfigOp(new ConfigEntry("delete.retention.ms", "0"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("min.cleanable.dirty.ratio", "0.0"), AlterConfigOp.OpType.SET)
Review Comment:
nit: Could we replace the names by their constants?
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,182 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedConsumerGroupAndTombstoneRemoved(): Unit = {
+ withAdmin { admin =>
+ TestUtils.createTopicWithAdminRaw(
+ admin = admin,
+ topic = "foo",
+ numPartitions = 3
+ )
+
+ // Create a classic group with one member and commit offsets.
+ withConsumer(groupId = "grp4", groupProtocol = GroupProtocol.CLASSIC, enableAutoCommit = false) { consumer =>
Review Comment:
Should we add a test case with a "simple classic" group too?
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2907409899
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
Ah. I missed that we are passing `f` as parameter... My bad. -- Might be worth to give it better name?
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2911141650
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,182 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedConsumerGroupAndTombstoneRemoved(): Unit = {
+ withAdmin { admin =>
+ TestUtils.createTopicWithAdminRaw(
+ admin = admin,
+ topic = "foo",
+ numPartitions = 3
+ )
+
+ // Create a classic group with one member and commit offsets.
+ withConsumer(groupId = "grp4", groupProtocol = GroupProtocol.CLASSIC, enableAutoCommit = false) { consumer =>
Review Comment:
Good idea, added two tests: `testCoordinatorFailoverAfterCompactingPartitionWithUpgradedSimpleConsumerGroup` and `testCoordinatorFailoverAfterCompactingPartitionWithUpgradedSimpleStreamsGroup`. These create a simple classic group by committing offsets directly (without subscribing), then upgrade to consumer/streams protocol. Since simple classic groups have no GroupMetadata records, there are no tombstones — the offset commit records always survive compaction, so a single compaction pass suffices.
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -331,12 +512,34 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def writeOneOffsetCommit(): Unit = {
+ // Write a single offset commit to create dirty data past the cleaner
+ // checkpoint so the cleaner will re-process previously compacted
+ // segments on the next compaction pass.
+ withConsumer(groupId = "compaction-trigger", groupProtocol = GroupProtocol.CLASSIC, enableAutoCommit = false) { consumer =>
+ val tp = new TopicPartition("foo", 0)
+ consumer.assign(java.util.List.of(tp))
+ consumer.commitSync(java.util.Map.of(tp, new OffsetAndMetadata(0)))
+ }
+ }
+
+ private def configureDeleteRetention(): Unit = {
+ withAdmin { admin =>
+ val resource = new ConfigResource(ConfigResource.Type.TOPIC, "__consumer_offsets")
+ admin.incrementalAlterConfigs(java.util.Map.of(resource, java.util.List.of(
+ new AlterConfigOp(new ConfigEntry("delete.retention.ms", "0"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("min.cleanable.dirty.ratio", "0.0"), AlterConfigOp.OpType.SET)
Review Comment:
Done, replaced with `TopicConfig.DELETE_RETENTION_MS_CONFIG`, `TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG`, and `Topic.GROUP_METADATA_TOPIC_NAME`.
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2905547837
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -366,6 +452,32 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ private def withStreamsApp(
+ applicationId: String,
+ inputTopic: String
+ )(f: KafkaStreams => Unit): Unit = {
+ val builder = new StreamsBuilder()
+ builder.stream(inputTopic)
+
+ val props = new Properties()
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId)
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, classOf[Serdes.StringSerde].getName)
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, StreamsGroupProtocol.STREAMS.name())
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "500")
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+
+ val streams = new KafkaStreams(builder.build(), props)
+ try {
+ streams.start()
+ f(streams)
Review Comment:
It's calling lambda "f" which is passed into this method on the Kafka Streams object. May look weird with Java eyes, but fairly idiomatic scala
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2892523518
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
Review Comment:
We call `close()` and with "streams" we always send a LEAVE_GROUP_REQUEST -- why would we need to wait for session timeout? -- Guess it's still ok to verify if the group is indeed empty, but the comment might be misleading?
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2896949235
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,79 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedStreamsGroup(): Unit = {
Review Comment:
This was an attempt to use the consumer test, but it does not fail if I remove the fix (and hte consumer test does not fail if I remove the consumer fix), so not sure whether to include it. The other one seems more comprehensive.
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2906982966
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,192 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedConsumerGroupAndTombstoneRemoved(): Unit = {
Review Comment:
@dajac I am adding a test for the "compaction and tombstone removal" scenario here also for consumer groups, as the existing integration test wasn't covering it. I suppose it's not necessary to create a separate ticket for that.
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2905559466
##########
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala:
##########
@@ -279,6 +284,79 @@ class GroupCoordinatorIntegrationTest(cluster: ClusterInstance) {
}
}
+ @ClusterTest(
+ types = Array(Type.KRAFT),
+ serverProperties = Array(
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
+ )
+ )
+ def testCoordinatorFailoverAfterCompactingPartitionWithUpgradedStreamsGroup(): Unit = {
Review Comment:
f(streams) calls the closure which waits for RUNNING state — this is sufficient for the group coordinator to have created the streams group records.
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2892523518
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
Review Comment:
Can't we change the code to actually send a LEAVE_GROUP_REQUEST on `close()?
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #21642:
URL: https://github.com/apache/kafka/pull/21642#issuecomment-4025290951
@mjsax Updated.
Turns out there were some changes required for both the consumer and the streams test to actually trigger the bug.
--
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-20254: Add integration tests for group migration with compaction and broker restart [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #21642:
URL: https://github.com/apache/kafka/pull/21642#issuecomment-4176843158
4.2: https://github.com/apache/kafka/commit/277663463c7143ceae348b3b87b2f63dc20b397a
--
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-20254: Add integration test for streams group migration with broker restart [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21642:
URL: https://github.com/apache/kafka/pull/21642#discussion_r2897926327
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RebalanceProtocolMigrationIntegrationTest.java:
##########
@@ -134,6 +145,125 @@ public void shouldMigrateToAndFromStreamsRebalanceProtocol() throws Exception {
processExactlyOneRecord(streamsBuilder, props, "3", "C");
}
+ @Test
+ public void shouldMigrateFromClassicToStreamsAfterBrokerRestart() throws Exception {
+ // This test reproduces KAFKA-20254: after log compaction removes the
+ // GroupMetadata tombstone from __consumer_offsets, offset commit records
+ // (which precede the streams group records in the log) create a simple
+ // classic group during replay, and then the streams group records must
+ // handle this existing simple classic group.
+ final StreamsBuilder streamsBuilder = new StreamsBuilder();
+ final KStream<String, String> input = streamsBuilder.stream(
+ inputTopic, Consumed.with(Serdes.String(), Serdes.String()));
+ input.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
+
+ final Properties props = props();
+ final String appId = props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ // Step 1: Run with the classic protocol and process a record.
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ processExactlyOneRecord(streamsBuilder, props, "1", "A");
+
+ // Wait for session to time out so the group becomes empty.
+ try (final Admin adminClient = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()))) {
+ waitForEmptyConsumerGroup(adminClient, appId, 1000);
+ }
+
+ // Step 2: Commit an offset for an "orphan" topic using the same group ID.
+ // This offset commit record will precede the streams group records in the
+ // log and will survive compaction because the streams group never commits
+ // for this topic-partition.
Review Comment:
I see -- quite a hack. But makes sense.
--
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