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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
