You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chickenchickenlove (via GitHub)" <gi...@apache.org> on 2026/06/13 03:06:35 UTC
[PR] KAFKA-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
chickenchickenlove opened a new pull request, #22561:
URL: https://github.com/apache/kafka/pull/22561
### Summary
This PR is based on the `KAFKA-20169-CLIENT-CORE` branch. (https://github.com/apache/kafka/pull/22559)
It adds a new ducktape system test to verify that Static Membership works correctly with the Streams rebalance protocol.
The new test covers the following behavior:
- A bounced static Streams member reuses its persisted `processId` after restart.
- Surviving static members are not forced into reconciliation when another static member temporarily bounces and rejoins.
- The test runs with the Streams group protocol and repeatedly bounces each static member to validate stable behavior across restarts.
### Changes
- Added `StaticMemberPersistentProcessIdTestClient`, a Streams test client with a persistent state store so the Streams `processId` is written to disk and reused after restart.
- Added a corresponding `StaticMemberPersistentProcessIdTestService` for ducktape.
- Added a new ducktape test: `StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation`
### Test Results
```
[INFO:2026-06-12 20:00:12,560]: Discovered 1 tests to run
[INFO:2026-06-12 20:00:12,561]: starting test run with session id 2026-06-12--005...
[INFO:2026-06-12 20:00:12,561]: running 1 tests...
[INFO:2026-06-12 20:00:12,561]: Triggering test 1 of 1...
[INFO:2026-06-12 20:00:12,566]: RunnerClient: Loading test {'directory': '/opt/kafka-dev/tests/kafkatest/tests/streams', 'file_name': 'streams_static_membership_streams_protocol_test.py', 'cls_name': 'StreamsStaticMembershipStreamsProtocolTest', 'method_name': 'test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation', 'injected_args': {'metadata_quorum': 'ISOLATED_KRAFT'}}
[INFO:2026-06-12 20:00:12,568]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: on run 1/1
[INFO:2026-06-12 20:00:12,569]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: Setting up...
[INFO:2026-06-12 20:00:12,569]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: Running...
[INFO:2026-06-12 20:01:42,279]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: Tearing down...
[INFO:2026-06-12 20:02:04,736]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: PASS
[INFO:2026-06-12 20:02:04,738]: RunnerClient: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT: Data: None
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.14.0
session_id: 2026-06-12--005
run time: 1 minute 52.289 seconds
tests run: 1
passed: 1
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id: kafkatest.tests.streams.streams_static_membership_streams_protocol_test.StreamsStaticMembershipStreamsProtocolTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 52.168 seconds
```
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4830205271
@chickenchickenlove Thanks a lot for the hard work!
Looking at the docs overall, I think there are a few things worth adding before this lands.
The most important one is the stateless topology limitation. `StateDirectory.initializeProcessId()` only persists the processId when `hasPersistentStores=true` — stateless topologies get a new random UUID on every restart. That means the thread-level instance IDs change on each restart, and the broker sees new members every time rather than the same static ones rejoining. Users who add `group.instance.id` to a stateless app will silently get the wrong behavior with no guidance anywhere in the docs. This deserves a note in the 4.4.0 upgrade guide section and in `streams-rebalance-protocol.md`.
Second, the existing static membership entry is in the `## Streams API changes in 4.3.0` section, but the implementation merged to trunk (4.4.0). It should move to the 4.4.0 section.
Third, the thread-scoped instance ID format isn't documented anywhere. Users need to know that `group.instance.id=my-app` causes threads to register as `my-app-1`, `my-app-2`, etc. — both so they can reason about uniqueness across instances, and so they understand what shows up in admin tools.
Fourth, `config-streams.md` has no dedicated entry for `group.instance.id`. Regular consumer static membership has a full entry in `consumer-configs.md` with semantics and caveats. For Streams it's only mentioned in passing in the `group.protocol` description. Since `group.instance.id` is a `ConsumerConfig` rather than a `StreamsConfig` key, this is easy to miss. A short dedicated section in the config doc or a cross-reference from `streams-rebalance-protocol.md` would help.
Finally, the existing `CloseOptions` note in the 3.3.0 upgrade guide section still says "only works for static consumer groups at this point" — that caveat should be lifted in the 4.4.0 entry now that streams static membership works.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4827534523
@lucasbru
Thank you so much for taking the time to review this PR over an extended period. 🙇♂️
Your thoughtful and proactive support helped me get this work across the finish 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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3474603672
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -53,12 +63,7 @@ def __init__(self, test_context):
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
Review Comment:
This test still uses the default classic group protocol through `create_processors(self.num_threads)`. I only enable the streams group protocol in the tests that explicitly cover streams-group behavior.
```python
def create_processors(self, num_threads, group_protocol="classic", persistent_process_id_store_enabled=False):
return [ ... ]
```
What do you think?
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3419014086
##########
tests/kafkatest/tests/streams/streams_static_membership_streams_protocol_test.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0.
+
+import re
+
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.tests.test import Test
+
+from kafkatest.services.kafka import KafkaService, quorum
+from kafkatest.services.streams import StaticMemberPersistentProcessIdTestService
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.tests.streams.utils import verify_running, verify_stopped, stop_processors
+
+
+class StreamsStaticMembershipStreamsProtocolTest(Test):
+ """
+ Streams group protocol specific static membership test.
+ This test verifies the following behavior under the streams group protocol:
+
+ 1. The bounced static member reuses the same persisted processId.
+ 2. Surviving members do not reconcile because of another member's temporary bounce.
+ """
+
+ input_topic = "inputTopic"
+ running_message = "REBALANCING -> RUNNING"
+ stopped_message = "Static membership persistent-process-id test closed"
+ processed_message = "PROCESSED"
+
+ num_threads = 3
+ num_bounces = 3
+ group_protocol = "streams"
+
+ initial_process_id_pattern = re.compile(r"No process id found on disk, got fresh process id ([0-9a-fA-F-]+)")
+ random_process_id_pattern = re.compile(r"Created new process id: ([0-9a-fA-F-]+)")
+ reused_process_id_pattern = re.compile(r"Reading UUID from process file: ([0-9a-fA-F-]+)")
+
+ def __init__(self, test_context):
+ super(StreamsStaticMembershipStreamsProtocolTest, self).__init__(test_context)
+ self.topics = {
+ self.input_topic: {"partitions": 18},
+ }
+
+ self.kafka = KafkaService(self.test_context, num_nodes=3,
+ zk=None, topics=self.topics, controller_num_nodes_override=1)
+
+ self.producer = VerifiableProducer(self.test_context,
+ 1,
+ self.kafka,
+ self.input_topic,
+ throughput=1000,
+ acks=1)
+
+ @cluster(num_nodes=8)
+ @matrix(metadata_quorum=[quorum.isolated_kraft])
+ def test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation(self, metadata_quorum):
+ self.kafka.start()
+
+ processor1 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-A", self.num_threads, self.group_protocol)
+ processor2 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-B", self.num_threads, self.group_protocol)
+ processor3 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-C", self.num_threads, self.group_protocol)
+
+ processors = [processor1, processor2, processor3]
+
+ self.producer.start()
+
+ for processor in processors:
+ processor.CLEAN_NODE_ENABLED = False
+ processor.INPUT_TOPIC = self.input_topic
+ verify_running(processor, self.running_message)
+
+ self.verify_processing(processors)
+
+ baseline_process_ids = {
+ processor: self.assert_initial_process_id_persisted(processor)
+ for processor in processors
+ }
+
+ for _ in range(self.num_bounces):
+ for bounced in processors:
+ checkpoints = {
+ processor: {
+ "log": self._line_count(processor, processor.LOG_FILE),
+ "stdout": self._line_count(processor, processor.STDOUT_FILE),
+ }
+ for processor in processors
+ }
+
+ verify_stopped(bounced, self.stopped_message)
+ verify_running(bounced, self.running_message)
+
+ self.assert_same_process_id_reused(bounced, checkpoints[bounced]["log"], baseline_process_ids[bounced])
+
+ for survivor in processors:
+ if survivor is bounced:
+ continue
+ self.assert_survivor_was_unaffected(survivor, checkpoints[survivor]["log"])
Review Comment:
`checkpoints` captures both log and stdout line counts, but only the log checkpoint is used. The extra stdout `awk` call adds SSH overhead inside the bounce loops without affecting assertions; simplifying this avoids unnecessary remote commands and makes the intent clearer.
##########
streams/src/test/java/org/apache/kafka/streams/tests/StaticMemberPersistentProcessIdTestClient.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streams.tests;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.Stores;
+
+import java.util.Objects;
+import java.util.Properties;
+
+public class StaticMemberPersistentProcessIdTestClient {
+
+ private static final String TEST_NAME = "StaticMemberPersistentProcessIdTestClient";
+
+ @SuppressWarnings("unchecked")
+ public static void main(final String[] args) throws Exception {
+ if (args.length < 1) {
+ System.err.println(TEST_NAME + " requires one argument (properties-file) but none provided: ");
+ }
+
+ System.out.println("StreamsTest instance started");
+
+ final String propFileName = args[0];
+ final Properties streamsProperties = Utils.loadProps(propFileName);
Review Comment:
The argument length check prints an error but then continues and unconditionally reads `args[0]`, which would throw `ArrayIndexOutOfBoundsException` if the class is ever invoked incorrectly (and can also proceed if `Exit.exit` is overridden in tests). Consider exiting early after reporting the error.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3473508928
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -97,6 +101,244 @@ def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self
self.producer.stop()
self.kafka.stop(timeout_sec=120)
+ @cluster(num_nodes=8)
+ @matrix(metadata_quorum=[quorum.isolated_kraft])
+ def test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation(self, metadata_quorum):
+ self.kafka.start()
+
+ processors = self.create_processors(
+ self.num_threads,
+ group_protocol=self.streams_group_protocol,
+ persistent_process_id_store_enabled=True
+ )
+
+ self.producer.start()
+
+ initial_log_checkpoints = {}
+ for processor in processors:
+ processor.CLEAN_NODE_ENABLED = False
+ self.set_topics(processor)
+ initial_log_checkpoints[processor] = self._line_count(processor, processor.LOG_FILE)
+ verify_running(processor, self.running_message)
+
+ self.verify_processing(processors)
+
+ baseline_process_ids = {
+ processor: self.assert_initial_process_id_persisted(processor, initial_log_checkpoints[processor])
+ for processor in processors
+ }
+
+ for _ in range(self.num_bounces):
+ for bounced in processors:
+ checkpoints = {
+ processor: self._line_count(processor, processor.LOG_FILE)
+ for processor in processors
+ }
+
+ verify_stopped(bounced, self.stopped_message)
+ verify_running(bounced, self.running_message)
+
+ self.assert_same_process_id_reused(
+ bounced,
+ checkpoints[bounced],
+ baseline_process_ids[bounced]
+ )
+
+ for survivor in processors:
+ if survivor is not bounced:
+ self.assert_survivor_was_unaffected(survivor, checkpoints[survivor])
+
+ self.verify_processing(processors)
+
+ stop_processors(processors, self.stopped_message)
+
+ self.producer.stop()
+ self.kafka.stop(timeout_sec=120)
+
+ @cluster(num_nodes=8)
+ @matrix(
+ bounce_mode=["all", "rolling"],
+ metadata_quorum=[quorum.isolated_kraft]
+ )
+ def test_static_member_process_id_persisted_after_rejoin(self, bounce_mode, metadata_quorum):
+ self.kafka.start()
+
+ processors = self.create_processors(
+ self.num_threads,
+ group_protocol=self.streams_group_protocol,
+ persistent_process_id_store_enabled=True
+ )
+
+ self.producer.start()
+
+ initial_log_checkpoints = {}
+ for processor in processors:
+ processor.CLEAN_NODE_ENABLED = False
+ self.set_topics(processor)
+ initial_log_checkpoints[processor] = self._line_count(processor, processor.LOG_FILE)
+ verify_running(processor, self.running_message)
+
+ self.verify_processing(processors)
+
+ baseline_process_ids = {
+ processor: self.assert_initial_process_id_persisted(processor, initial_log_checkpoints[processor])
+ for processor in processors
+ }
+
+ if bounce_mode == "all":
+ checkpoints = {
+ processor: self._line_count(processor, processor.LOG_FILE)
+ for processor in processors
+ }
+
+ for processor in processors:
+ verify_stopped(processor, self.stopped_message)
+
+ for processor in processors:
+ verify_running(processor, self.running_message)
+
+ for processor in processors:
+ self.assert_same_process_id_reused(
+ processor,
+ checkpoints[processor],
+ baseline_process_ids[processor]
+ )
+ else:
+ for processor in processors:
+ checkpoint = self._line_count(processor, processor.LOG_FILE)
+
+ verify_stopped(processor, self.stopped_message)
+ verify_running(processor, self.running_message)
+
+ self.assert_same_process_id_reused(
+ processor,
+ checkpoint,
+ baseline_process_ids[processor]
+ )
+
+ self.verify_processing(processors)
+
+ for node in self.kafka.nodes:
+ self.kafka.restart_node(node, clean_shutdown=True, timeout_sec=120)
+ assert not self.fenced_processors(processors), (
+ "Static Streams member unexpectedly failed after broker rolling bounce"
+ )
+ self.verify_processing(processors)
+
+ stop_processors(processors, self.stopped_message)
+
+ self.producer.stop()
+ self.kafka.stop(timeout_sec=120)
+
+ @cluster(num_nodes=9)
+ @matrix(
+ fencing_stage=["stable", "all"],
+ metadata_quorum=[quorum.isolated_kraft]
+ )
+ def test_fencing_static_streams_member(self, fencing_stage, metadata_quorum):
+ self.kafka.start()
+
+ processors = self.create_processors(
+ self.num_threads,
+ group_protocol=self.streams_group_protocol
+ )
+ conflict_processor = StaticMemberTestService(
+ self.test_context,
+ self.kafka,
+ processors[0].GROUP_INSTANCE_ID,
+ self.num_threads,
+ self.streams_group_protocol
+ )
+
+ self.producer.start()
+
+ if fencing_stage == "stable":
+ for processor in processors:
+ self.set_topics(processor)
+ verify_running(processor, self.running_message)
+
+ self.verify_processing(processors)
+
+ checkpoints = {
+ processor: self._line_count(processor, processor.LOG_FILE)
+ for processor in processors
+ }
+
+ self.set_topics(conflict_processor)
+ with conflict_processor.node.account.monitor_log(conflict_processor.LOG_FILE) as monitor:
+ conflict_processor.start()
+ monitor.wait_until(
+ "terminal ERROR state",
+ timeout_sec=60,
+ err_msg="Never saw the conflicting static Streams member fail"
+ )
+
+ conflict_processor.wait(timeout_sec=60)
+
+ for processor in processors:
+ self.assert_survivor_was_unaffected(processor, checkpoints[processor])
+
+ self.verify_processing(processors)
+
+ stop_processors(processors, self.stopped_message)
+
+ verify_running(conflict_processor, self.running_message)
+ self.verify_processing([conflict_processor])
+ verify_stopped(conflict_processor, self.stopped_message)
+ else:
+ duplicate_processors = [processors[0], conflict_processor]
+ all_processors = duplicate_processors + processors[1:]
+
+ for processor in all_processors:
+ self.set_topics(processor)
+ processor.start()
+
+ wait_until(
+ lambda: len(self.fenced_processors(duplicate_processors)) >= 1,
+ timeout_sec=60,
+ err_msg="Timed out waiting for one duplicate static Streams member to fail"
+ )
+
+ fenced_processors = self.fenced_processors(duplicate_processors)
+ assert len(fenced_processors) == 1, (
+ "Expected exactly one duplicate static Streams member to fail, but saw %d"
+ % len(fenced_processors)
+ )
+
+ for processor in fenced_processors:
+ processor.wait(timeout_sec=60)
+
+ active_processors = [
+ processor for processor in processors + [conflict_processor]
+ if processor not in fenced_processors
+ ]
+
+ for processor in active_processors:
+ self.wait_for_file_contains(
+ processor,
+ processor.STDOUT_FILE,
+ self.running_message,
+ "Never saw running state for active static Streams member %s"
+ % processor.GROUP_INSTANCE_ID
+ )
+
+ self.verify_processing(active_processors)
+
+ stop_processors(active_processors, self.stopped_message)
+
+ self.producer.stop()
+ self.kafka.stop(timeout_sec=120)
+
+ def create_processors(self, num_threads, group_protocol="classic", persistent_process_id_store_enabled=False):
+ return [
+ StaticMemberTestService(self.test_context, self.kafka, "consumer-A", num_threads, group_protocol,
+ persistent_process_id_store_enabled),
+ StaticMemberTestService(self.test_context, self.kafka, "consumer-B", num_threads, group_protocol,
+ persistent_process_id_store_enabled),
+ StaticMemberTestService(self.test_context, self.kafka, "consumer-C", num_threads, group_protocol,
+ persistent_process_id_store_enabled)
+ ]
+
def verify_processing(self, processors):
Review Comment:
_line_count + _read_lines_since reimplements what ducktape's node.account.monitor_log() already provides. Starting a monitor just before each bounce and checking within that window would be the standard ducktape idiom and removes ~20 lines of custom SSH/sed infrastructure.
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3420913812
##########
tests/kafkatest/tests/streams/streams_static_membership_streams_protocol_test.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
Review Comment:
This is largely a copy of the existing static membership test infra (StreamsStaticMembershipTest, StaticMemberTestService, StaticMemberTestClient). The only real new bits are the persistent store so the processId survives restart and group.protocol=streams. Could we instead adapt the existing test/service rather than forking new copies - e.g. parameterize StaticMemberTestService with group_protocol and add the persistent store to the existing client? The new assertion (survivor non-reconciliation / processId reuse) can be a new test method on the existing class.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3474550739
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsStaticMemberIntegrationTest.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(600)
+public class KafkaStreamsStaticMemberIntegrationTest {
+
+ private static MockTime mockTime;
+
+ protected static final String INPUT_TOPIC = "inputTopic";
+ protected static final String OUTPUT_TOPIC = "outputTopic";
+ protected static final String GROUP_INSTANCE_ID = "someGroupInstance";
+
+ protected Properties streamsConfig;
+ protected static KafkaStreams streams;
+ protected static Admin adminClient;
+ protected Properties commonClientConfig;
+ private Properties producerConfig;
+ protected Properties resultConsumerConfig;
+ private final File testFolder = TestUtils.tempDirectory();
+
+ public static final EmbeddedKafkaCluster CLUSTER;
+
+ static {
+ final Properties brokerProps = new Properties();
+ brokerProps.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ CLUSTER = new EmbeddedKafkaCluster(1, brokerProps);
+ }
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ Utils.closeQuietly(adminClient, "admin");
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testName) throws Exception {
+ mockTime = CLUSTER.time;
+
+ final String appID = safeUniqueTestName(testName);
+
+ commonClientConfig = new Properties();
+ commonClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
+ // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
+ streamsConfig.putAll(commonClientConfig);
+
+ producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.putAll(commonClientConfig);
+
+ resultConsumerConfig = new Properties();
+ resultConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
+ resultConsumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ resultConsumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+ resultConsumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ resultConsumerConfig.putAll(commonClientConfig);
+
+ if (adminClient == null) {
+ adminClient = Admin.create(commonClientConfig);
+ }
+
+ CLUSTER.deleteAllTopics();
+ CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+ CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+
+ add10InputElements();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams = null;
+ }
+ Utils.delete(testFolder);
+ }
+
+ @Test
+ public void testStaticMemberJoinRegistersThreadScopedInstanceIdsStreamsProtocol() throws Exception {
+ final int numStreamThreads = 2;
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == numStreamThreads &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith(GROUP_INSTANCE_ID + "-"))
+ .isPresent() &&
+ member.memberEpoch() >= 0);
+ }, "Static streams members did not join with thread-scoped instance ids");
+ }
+
+ @Test
+ public void testStaticMemberCloseWithLeaveGroupTriggersRebalanceStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ final AtomicInteger secondStreamsRebalances = new AtomicInteger(0);
+
+ try (final KafkaStreams streams1 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(firstInstanceId, firstStateDir));
+ final KafkaStreams streams2 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(secondInstanceId, secondStateDir))) {
+ streams2.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.REBALANCING) {
+ secondStreamsRebalances.incrementAndGet();
+ }
+ });
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(streams1, streams2));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 2;
+ }, "Streams group did not reach 2 static members");
+
+ final int initialStreams2Rebalances = secondStreamsRebalances.get();
+
+ streams1.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 1 &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith("instance-2-"))
+ .isPresent());
+ }, "Streams group did not retain only the remaining static member after LEAVE_GROUP close");
+
+ TestUtils.waitForCondition(
+ () -> secondStreamsRebalances.get() > initialStreams2Rebalances,
+ "Remaining streams instance did not observe a rebalance after static member left the group"
+ );
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterRemainInGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ final Optional<String> rejoinedMemberId = staticStreamsMemberId(groupDescription, firstInstanceId);
+
+ return groupDescription.members().size() == 2 &&
+ rejoinedMemberId.isPresent() &&
+ !rejoinedMemberId.get().equals(firstMemberId) &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin with a new member id after REMAIN_IN_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterLeaveGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not leave the group with LEAVE_GROUP");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin after LEAVE_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberRemainingInGroupExpiresAndTriggersRebalanceStreamsProtocol() throws Exception {
+ final int sessionTimeoutMs = 1000;
+ final int heartbeatIntervalMs = 100;
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ CLUSTER.setGroupHeartbeatTimeout(applicationId, heartbeatIntervalMs);
+ CLUSTER.setGroupSessionTimeout(applicationId, sessionTimeoutMs);
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ final int groupEpochAfterStaticLeave = describeStreamsGroup(applicationId).groupEpoch();
+
+ TestUtils.waitForCondition(() -> {
+ mockTime.sleep(heartbeatIntervalMs);
+
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ groupDescription.groupEpoch() > groupEpochAfterStaticLeave &&
+ staticStreamsMemberId(groupDescription, firstInstanceId).isEmpty() &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member was not removed after session timeout");
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testDuplicateStaticInstanceIdDoesNotReplaceExistingStreamsMember() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
Review Comment:
Please refer to https://github.com/apache/kafka/pull/22561#discussion_r3474548798
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4763967653
@lucasbru
Thanks for the review.
I've addressed it based on your comments and test results are here!
```shell
====================================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.14.0
session_id: 2026-06-21--008
run time: 3 minutes 38.640 seconds
tests run: 2
passed: 2
flaky: 0
failed: 0
ignored: 0
====================================================================================================
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_rolling_bounces_will_not_trigger_rebalance_under_static_membership.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 37.775 seconds
----------------------------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 2 minutes 0.611 seconds
----------------------------------------------------------------------------------------------------
```
When you get a chance, please take another look. 🙇♂️
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4799902929
@lucasbru
Thanks for the review.
I've addressed it based on your comments.
Also, I wrote the reply to your comments!
When you get a chance, please take a look. 🙇♂️
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3474690593
##########
tests/kafkatest/tests/streams/streams_static_membership_streams_protocol_test.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0.
+
+import re
+
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.tests.test import Test
+
+from kafkatest.services.kafka import KafkaService, quorum
+from kafkatest.services.streams import StaticMemberPersistentProcessIdTestService
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.tests.streams.utils import verify_running, verify_stopped, stop_processors
+
+
+class StreamsStaticMembershipStreamsProtocolTest(Test):
+ """
+ Streams group protocol specific static membership test.
+ This test verifies the following behavior under the streams group protocol:
+
+ 1. The bounced static member reuses the same persisted processId.
+ 2. Surviving members do not reconcile because of another member's temporary bounce.
+ """
+
+ input_topic = "inputTopic"
+ running_message = "REBALANCING -> RUNNING"
+ stopped_message = "Static membership persistent-process-id test closed"
+ processed_message = "PROCESSED"
+
+ num_threads = 3
+ num_bounces = 3
+ group_protocol = "streams"
+
+ initial_process_id_pattern = re.compile(r"No process id found on disk, got fresh process id ([0-9a-fA-F-]+)")
+ random_process_id_pattern = re.compile(r"Created new process id: ([0-9a-fA-F-]+)")
+ reused_process_id_pattern = re.compile(r"Reading UUID from process file: ([0-9a-fA-F-]+)")
+
+ def __init__(self, test_context):
+ super(StreamsStaticMembershipStreamsProtocolTest, self).__init__(test_context)
+ self.topics = {
+ self.input_topic: {"partitions": 18},
+ }
+
+ self.kafka = KafkaService(self.test_context, num_nodes=3,
+ zk=None, topics=self.topics, controller_num_nodes_override=1)
+
+ self.producer = VerifiableProducer(self.test_context,
+ 1,
+ self.kafka,
+ self.input_topic,
+ throughput=1000,
+ acks=1)
+
+ @cluster(num_nodes=8)
+ @matrix(metadata_quorum=[quorum.isolated_kraft])
+ def test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation(self, metadata_quorum):
+ self.kafka.start()
+
+ processor1 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-A", self.num_threads, self.group_protocol)
+ processor2 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-B", self.num_threads, self.group_protocol)
+ processor3 = StaticMemberPersistentProcessIdTestService(self.test_context, self.kafka, "consumer-C", self.num_threads, self.group_protocol)
+
+ processors = [processor1, processor2, processor3]
+
+ self.producer.start()
+
+ for processor in processors:
+ processor.CLEAN_NODE_ENABLED = False
+ processor.INPUT_TOPIC = self.input_topic
+ verify_running(processor, self.running_message)
+
+ self.verify_processing(processors)
+
+ baseline_process_ids = {
+ processor: self.assert_initial_process_id_persisted(processor)
+ for processor in processors
+ }
+
+ for _ in range(self.num_bounces):
+ for bounced in processors:
+ checkpoints = {
+ processor: {
+ "log": self._line_count(processor, processor.LOG_FILE),
+ "stdout": self._line_count(processor, processor.STDOUT_FILE),
+ }
+ for processor in processors
+ }
+
+ verify_stopped(bounced, self.stopped_message)
+ verify_running(bounced, self.running_message)
+
+ self.assert_same_process_id_reused(bounced, checkpoints[bounced]["log"], baseline_process_ids[bounced])
+
+ for survivor in processors:
+ if survivor is bounced:
+ continue
+ self.assert_survivor_was_unaffected(survivor, checkpoints[survivor]["log"])
Review Comment:
Outdated.
##########
streams/src/test/java/org/apache/kafka/streams/tests/StaticMemberPersistentProcessIdTestClient.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streams.tests;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.internals.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.Stores;
+
+import java.util.Objects;
+import java.util.Properties;
+
+public class StaticMemberPersistentProcessIdTestClient {
+
+ private static final String TEST_NAME = "StaticMemberPersistentProcessIdTestClient";
+
+ @SuppressWarnings("unchecked")
+ public static void main(final String[] args) throws Exception {
+ if (args.length < 1) {
+ System.err.println(TEST_NAME + " requires one argument (properties-file) but none provided: ");
+ }
+
+ System.out.println("StreamsTest instance started");
+
+ final String propFileName = args[0];
+ final Properties streamsProperties = Utils.loadProps(propFileName);
Review Comment:
Outdated.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3477655914
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -53,12 +63,7 @@ def __init__(self, test_context):
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
Review Comment:
@lucasbru
Ah, My bad... 😢
I've added it and tested them. (c33cc98a69b5b5509f20bcad0292f437288d3ad2)
Sorry to bother you.
When you get a chance, please take another look. 🙇♂️
```
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.14.0
session_id: 2026-06-25--003
run time: 3 minutes 46.611 seconds
tests run: 2
passed: 2
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_rolling_bounces_will_not_trigger_rebalance_under_static_membership.group_protocol=classic.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 39.153 seconds
--------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_rolling_bounces_will_not_trigger_rebalance_under_static_membership.group_protocol=streams.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 2 minutes 7.213 seconds
--------------------------------------------------------------------------------
```
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4788258009
@lucasbru
I've added some duck tape test cases.
When you get a chance, please take a look. 🙇♂️
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3474548798
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsStaticMemberIntegrationTest.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(600)
+public class KafkaStreamsStaticMemberIntegrationTest {
+
+ private static MockTime mockTime;
+
+ protected static final String INPUT_TOPIC = "inputTopic";
+ protected static final String OUTPUT_TOPIC = "outputTopic";
+ protected static final String GROUP_INSTANCE_ID = "someGroupInstance";
+
+ protected Properties streamsConfig;
+ protected static KafkaStreams streams;
+ protected static Admin adminClient;
+ protected Properties commonClientConfig;
+ private Properties producerConfig;
+ protected Properties resultConsumerConfig;
+ private final File testFolder = TestUtils.tempDirectory();
+
+ public static final EmbeddedKafkaCluster CLUSTER;
+
+ static {
+ final Properties brokerProps = new Properties();
+ brokerProps.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ CLUSTER = new EmbeddedKafkaCluster(1, brokerProps);
+ }
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ Utils.closeQuietly(adminClient, "admin");
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testName) throws Exception {
+ mockTime = CLUSTER.time;
+
+ final String appID = safeUniqueTestName(testName);
+
+ commonClientConfig = new Properties();
+ commonClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
+ // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
+ streamsConfig.putAll(commonClientConfig);
+
+ producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.putAll(commonClientConfig);
+
+ resultConsumerConfig = new Properties();
+ resultConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
+ resultConsumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ resultConsumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+ resultConsumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ resultConsumerConfig.putAll(commonClientConfig);
+
+ if (adminClient == null) {
+ adminClient = Admin.create(commonClientConfig);
+ }
+
+ CLUSTER.deleteAllTopics();
+ CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+ CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+
+ add10InputElements();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams = null;
+ }
+ Utils.delete(testFolder);
+ }
+
+ @Test
+ public void testStaticMemberJoinRegistersThreadScopedInstanceIdsStreamsProtocol() throws Exception {
+ final int numStreamThreads = 2;
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == numStreamThreads &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith(GROUP_INSTANCE_ID + "-"))
+ .isPresent() &&
+ member.memberEpoch() >= 0);
+ }, "Static streams members did not join with thread-scoped instance ids");
+ }
+
+ @Test
+ public void testStaticMemberCloseWithLeaveGroupTriggersRebalanceStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ final AtomicInteger secondStreamsRebalances = new AtomicInteger(0);
+
+ try (final KafkaStreams streams1 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(firstInstanceId, firstStateDir));
+ final KafkaStreams streams2 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(secondInstanceId, secondStateDir))) {
+ streams2.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.REBALANCING) {
+ secondStreamsRebalances.incrementAndGet();
+ }
+ });
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(streams1, streams2));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 2;
+ }, "Streams group did not reach 2 static members");
+
+ final int initialStreams2Rebalances = secondStreamsRebalances.get();
+
+ streams1.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 1 &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith("instance-2-"))
+ .isPresent());
+ }, "Streams group did not retain only the remaining static member after LEAVE_GROUP close");
+
+ TestUtils.waitForCondition(
+ () -> secondStreamsRebalances.get() > initialStreams2Rebalances,
+ "Remaining streams instance did not observe a rebalance after static member left the group"
+ );
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterRemainInGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ final Optional<String> rejoinedMemberId = staticStreamsMemberId(groupDescription, firstInstanceId);
+
+ return groupDescription.members().size() == 2 &&
+ rejoinedMemberId.isPresent() &&
+ !rejoinedMemberId.get().equals(firstMemberId) &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin with a new member id after REMAIN_IN_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterLeaveGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not leave the group with LEAVE_GROUP");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin after LEAVE_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberRemainingInGroupExpiresAndTriggersRebalanceStreamsProtocol() throws Exception {
+ final int sessionTimeoutMs = 1000;
Review Comment:
@lucasbru
Thanks for the comments. That makes sense to me. 👍
I looked into this, but I don’t think we can reliably cover this with the current in-process `KafkaStreams` integration test setup.
For the active timeout case, if we stop a Streams instance normally, it sends the static leave heartbeat first, so it becomes the soft-leave path this test already covers. If we restart it, it joins as a new member. To test a member that just stops heartbeating while still active, we would need to simulate a crash or network partition.
The same applies to the fenced stale-member case: we would need the old member to come back and heartbeat with its old identity after another member has replaced it. That is hard to do cleanly from this integration test without a fault-injection hook or a separate process-based test, which would likely be brittle.
I think this behavior is better covered at the coordinator / heartbeat-manager level, where we can directly create those protocol states without relying on timing or process failures.
Alternatively, we could test this by sending StreamsGroupHeartbeat requests directly, but that would not really be an integration test in the usual sense. It would also be semantically very similar to the existing `GroupMetadataManagerTest` coverage.
What do you think? 🤔
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3488418953
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsStaticMemberIntegrationTest.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(600)
+public class KafkaStreamsStaticMemberIntegrationTest {
+
+ private static MockTime mockTime;
+
+ protected static final String INPUT_TOPIC = "inputTopic";
+ protected static final String OUTPUT_TOPIC = "outputTopic";
+ protected static final String GROUP_INSTANCE_ID = "someGroupInstance";
+
+ protected Properties streamsConfig;
+ protected static KafkaStreams streams;
+ protected static Admin adminClient;
+ protected Properties commonClientConfig;
+ private Properties producerConfig;
+ protected Properties resultConsumerConfig;
+ private final File testFolder = TestUtils.tempDirectory();
+
+ public static final EmbeddedKafkaCluster CLUSTER;
+
+ static {
+ final Properties brokerProps = new Properties();
+ brokerProps.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ CLUSTER = new EmbeddedKafkaCluster(1, brokerProps);
+ }
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ Utils.closeQuietly(adminClient, "admin");
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testName) throws Exception {
+ mockTime = CLUSTER.time;
+
+ final String appID = safeUniqueTestName(testName);
+
+ commonClientConfig = new Properties();
+ commonClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
+ // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
+ streamsConfig.putAll(commonClientConfig);
+
+ producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.putAll(commonClientConfig);
+
+ resultConsumerConfig = new Properties();
+ resultConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
+ resultConsumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ resultConsumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+ resultConsumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ resultConsumerConfig.putAll(commonClientConfig);
+
+ if (adminClient == null) {
+ adminClient = Admin.create(commonClientConfig);
+ }
+
+ CLUSTER.deleteAllTopics();
+ CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+ CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+
+ add10InputElements();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams = null;
+ }
+ Utils.delete(testFolder);
+ }
+
+ @Test
+ public void testStaticMemberJoinRegistersThreadScopedInstanceIdsStreamsProtocol() throws Exception {
+ final int numStreamThreads = 2;
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == numStreamThreads &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith(GROUP_INSTANCE_ID + "-"))
+ .isPresent() &&
+ member.memberEpoch() >= 0);
+ }, "Static streams members did not join with thread-scoped instance ids");
+ }
+
+ @Test
+ public void testStaticMemberCloseWithLeaveGroupTriggersRebalanceStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ final AtomicInteger secondStreamsRebalances = new AtomicInteger(0);
+
+ try (final KafkaStreams streams1 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(firstInstanceId, firstStateDir));
+ final KafkaStreams streams2 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(secondInstanceId, secondStateDir))) {
+ streams2.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.REBALANCING) {
+ secondStreamsRebalances.incrementAndGet();
+ }
+ });
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(streams1, streams2));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 2;
+ }, "Streams group did not reach 2 static members");
+
+ final int initialStreams2Rebalances = secondStreamsRebalances.get();
+
+ streams1.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 1 &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith("instance-2-"))
+ .isPresent());
+ }, "Streams group did not retain only the remaining static member after LEAVE_GROUP close");
+
+ TestUtils.waitForCondition(
+ () -> secondStreamsRebalances.get() > initialStreams2Rebalances,
+ "Remaining streams instance did not observe a rebalance after static member left the group"
+ );
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterRemainInGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ final Optional<String> rejoinedMemberId = staticStreamsMemberId(groupDescription, firstInstanceId);
+
+ return groupDescription.members().size() == 2 &&
+ rejoinedMemberId.isPresent() &&
+ !rejoinedMemberId.get().equals(firstMemberId) &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin with a new member id after REMAIN_IN_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterLeaveGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not leave the group with LEAVE_GROUP");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin after LEAVE_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberRemainingInGroupExpiresAndTriggersRebalanceStreamsProtocol() throws Exception {
+ final int sessionTimeoutMs = 1000;
Review Comment:
Yes, your reasoning makes sense to me. I suppose the coverage we have is pretty good already.
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3469716135
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java:
##########
@@ -1450,57 +1454,65 @@ public void testOnHeartbeatRequestSkippedWhenInLeaving() {
@Test
public void testLeaveGroupEpochIsStaticMemberEpochForStaticMember() {
- final StreamsMembershipManager staticMember = new StreamsMembershipManager(
- GROUP_ID,
- Optional.of("instance-1"),
- streamsRebalanceData, subscriptionState, backgroundEventHandler,
- new LogContext("test"), time, new Metrics(time)
- );
- assertEquals(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH, staticMember.leaveGroupEpoch());
+ try (final Metrics localMetrics = new Metrics(time)) {
+ final StreamsMembershipManager staticMember = new StreamsMembershipManager(
+ GROUP_ID,
+ Optional.of(INSTANCE_ID),
+ streamsRebalanceData, subscriptionState, backgroundEventHandler,
+ new LogContext("test"), time, localMetrics
+ );
+ assertEquals(StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH, staticMember.leaveGroupEpoch());
+ }
}
@Test
public void testLeaveGroupEpochIsDynamicMemberEpochForStaticMemberWithLeaveGroupOperation() {
- final StreamsMembershipManager staticMember = new StreamsMembershipManager(
Review Comment:
This and testLeaveGroupEpochIsStaticMemberEpochForStaticMemberWithRemainInGroup below are both covered by the parameterized testStaticMemberUsesExpectedLeaveEpochOnClose — they can be removed.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsStaticMemberIntegrationTest.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(600)
+public class KafkaStreamsStaticMemberIntegrationTest {
+
+ private static MockTime mockTime;
+
+ protected static final String INPUT_TOPIC = "inputTopic";
+ protected static final String OUTPUT_TOPIC = "outputTopic";
+ protected static final String GROUP_INSTANCE_ID = "someGroupInstance";
+
+ protected Properties streamsConfig;
+ protected static KafkaStreams streams;
+ protected static Admin adminClient;
+ protected Properties commonClientConfig;
+ private Properties producerConfig;
+ protected Properties resultConsumerConfig;
+ private final File testFolder = TestUtils.tempDirectory();
+
+ public static final EmbeddedKafkaCluster CLUSTER;
+
+ static {
+ final Properties brokerProps = new Properties();
+ brokerProps.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ CLUSTER = new EmbeddedKafkaCluster(1, brokerProps);
+ }
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ Utils.closeQuietly(adminClient, "admin");
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testName) throws Exception {
+ mockTime = CLUSTER.time;
+
+ final String appID = safeUniqueTestName(testName);
+
+ commonClientConfig = new Properties();
+ commonClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
+ // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
+ streamsConfig.putAll(commonClientConfig);
+
+ producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.putAll(commonClientConfig);
+
+ resultConsumerConfig = new Properties();
+ resultConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
+ resultConsumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ resultConsumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+ resultConsumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ resultConsumerConfig.putAll(commonClientConfig);
+
+ if (adminClient == null) {
+ adminClient = Admin.create(commonClientConfig);
+ }
+
+ CLUSTER.deleteAllTopics();
+ CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+ CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+
+ add10InputElements();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams = null;
+ }
+ Utils.delete(testFolder);
+ }
+
+ @Test
+ public void testStaticMemberJoinRegistersThreadScopedInstanceIdsStreamsProtocol() throws Exception {
+ final int numStreamThreads = 2;
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == numStreamThreads &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith(GROUP_INSTANCE_ID + "-"))
+ .isPresent() &&
+ member.memberEpoch() >= 0);
+ }, "Static streams members did not join with thread-scoped instance ids");
+ }
+
+ @Test
+ public void testStaticMemberCloseWithLeaveGroupTriggersRebalanceStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ final AtomicInteger secondStreamsRebalances = new AtomicInteger(0);
+
+ try (final KafkaStreams streams1 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(firstInstanceId, firstStateDir));
+ final KafkaStreams streams2 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(secondInstanceId, secondStateDir))) {
+ streams2.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.REBALANCING) {
+ secondStreamsRebalances.incrementAndGet();
+ }
+ });
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(streams1, streams2));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 2;
+ }, "Streams group did not reach 2 static members");
+
+ final int initialStreams2Rebalances = secondStreamsRebalances.get();
+
+ streams1.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 1 &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith("instance-2-"))
+ .isPresent());
+ }, "Streams group did not retain only the remaining static member after LEAVE_GROUP close");
+
+ TestUtils.waitForCondition(
+ () -> secondStreamsRebalances.get() > initialStreams2Rebalances,
+ "Remaining streams instance did not observe a rebalance after static member left the group"
+ );
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterRemainInGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ final Optional<String> rejoinedMemberId = staticStreamsMemberId(groupDescription, firstInstanceId);
+
+ return groupDescription.members().size() == 2 &&
+ rejoinedMemberId.isPresent() &&
+ !rejoinedMemberId.get().equals(firstMemberId) &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin with a new member id after REMAIN_IN_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterLeaveGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not leave the group with LEAVE_GROUP");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin after LEAVE_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberRemainingInGroupExpiresAndTriggersRebalanceStreamsProtocol() throws Exception {
+ final int sessionTimeoutMs = 1000;
Review Comment:
This only tests the soft-leave (REMAIN_IN_GROUP) -> session timeout path. Worth adding a test for a static member that simply times out while actively heartbeating (epoch > 0, no prior soft-leave).
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsStaticMemberIntegrationTest.java:
##########
@@ -0,0 +1,596 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.GroupProtocol;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("integration")
+@Timeout(600)
+public class KafkaStreamsStaticMemberIntegrationTest {
+
+ private static MockTime mockTime;
+
+ protected static final String INPUT_TOPIC = "inputTopic";
+ protected static final String OUTPUT_TOPIC = "outputTopic";
+ protected static final String GROUP_INSTANCE_ID = "someGroupInstance";
+
+ protected Properties streamsConfig;
+ protected static KafkaStreams streams;
+ protected static Admin adminClient;
+ protected Properties commonClientConfig;
+ private Properties producerConfig;
+ protected Properties resultConsumerConfig;
+ private final File testFolder = TestUtils.tempDirectory();
+
+ public static final EmbeddedKafkaCluster CLUSTER;
+
+ static {
+ final Properties brokerProps = new Properties();
+ brokerProps.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
+ CLUSTER = new EmbeddedKafkaCluster(1, brokerProps);
+ }
+
+ @BeforeAll
+ public static void startCluster() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void closeCluster() {
+ Utils.closeQuietly(adminClient, "admin");
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void before(final TestInfo testName) throws Exception {
+ mockTime = CLUSTER.time;
+
+ final String appID = safeUniqueTestName(testName);
+
+ commonClientConfig = new Properties();
+ commonClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+
+ streamsConfig = new Properties();
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+ streamsConfig.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getPath());
+ streamsConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+ streamsConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
+ streamsConfig.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
+ // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
+ streamsConfig.putAll(commonClientConfig);
+
+ producerConfig = new Properties();
+ producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
+ producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+ producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ producerConfig.putAll(commonClientConfig);
+
+ resultConsumerConfig = new Properties();
+ resultConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, appID + "-result-consumer");
+ resultConsumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ resultConsumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
+ resultConsumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ resultConsumerConfig.putAll(commonClientConfig);
+
+ if (adminClient == null) {
+ adminClient = Admin.create(commonClientConfig);
+ }
+
+ CLUSTER.deleteAllTopics();
+ CLUSTER.createTopic(INPUT_TOPIC, 2, 1);
+ CLUSTER.createTopic(OUTPUT_TOPIC, 2, 1);
+
+ add10InputElements();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ if (streams != null) {
+ streams.close(Duration.ofSeconds(30));
+ streams = null;
+ }
+ Utils.delete(testFolder);
+ }
+
+ @Test
+ public void testStaticMemberJoinRegistersThreadScopedInstanceIdsStreamsProtocol() throws Exception {
+ final int numStreamThreads = 2;
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
+ streamsConfig.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, GROUP_INSTANCE_ID);
+
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == numStreamThreads &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith(GROUP_INSTANCE_ID + "-"))
+ .isPresent() &&
+ member.memberEpoch() >= 0);
+ }, "Static streams members did not join with thread-scoped instance ids");
+ }
+
+ @Test
+ public void testStaticMemberCloseWithLeaveGroupTriggersRebalanceStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ final AtomicInteger secondStreamsRebalances = new AtomicInteger(0);
+
+ try (final KafkaStreams streams1 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(firstInstanceId, firstStateDir));
+ final KafkaStreams streams2 = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), staticStreamsConfig(secondInstanceId, secondStateDir))) {
+ streams2.setStateListener((newState, oldState) -> {
+ if (newState == KafkaStreams.State.REBALANCING) {
+ secondStreamsRebalances.incrementAndGet();
+ }
+ });
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(streams1, streams2));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 2;
+ }, "Streams group did not reach 2 static members");
+
+ final int initialStreams2Rebalances = secondStreamsRebalances.get();
+
+ streams1.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription =
+ adminClient.describeStreamsGroups(Collections.singletonList(applicationId))
+ .describedGroups()
+ .get(applicationId)
+ .get();
+
+ return groupDescription.members().size() == 1 &&
+ groupDescription.members().stream().allMatch(member ->
+ member.instanceId()
+ .filter(instanceId -> instanceId.startsWith("instance-2-"))
+ .isPresent());
+ }, "Streams group did not retain only the remaining static member after LEAVE_GROUP close");
+
+ TestUtils.waitForCondition(
+ () -> secondStreamsRebalances.get() > initialStreams2Rebalances,
+ "Remaining streams instance did not observe a rebalance after static member left the group"
+ );
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterRemainInGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ final Optional<String> rejoinedMemberId = staticStreamsMemberId(groupDescription, firstInstanceId);
+
+ return groupDescription.members().size() == 2 &&
+ rejoinedMemberId.isPresent() &&
+ !rejoinedMemberId.get().equals(firstMemberId) &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin with a new member id after REMAIN_IN_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberCanRejoinAfterLeaveGroupStreamsProtocol() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final File restartedStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not leave the group with LEAVE_GROUP");
+
+ try (final KafkaStreams restartedStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, restartedStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(restartedStreams);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not rejoin after LEAVE_GROUP close");
+ }
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ Utils.delete(restartedStateDir);
+ }
+ }
+
+ @Test
+ public void testStaticMemberRemainingInGroupExpiresAndTriggersRebalanceStreamsProtocol() throws Exception {
+ final int sessionTimeoutMs = 1000;
+ final int heartbeatIntervalMs = 100;
+ final File firstStateDir = TestUtils.tempDirectory();
+ final File secondStateDir = TestUtils.tempDirectory();
+ final String applicationId = streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG);
+ final String firstInstanceId = "instance-1";
+ final String secondInstanceId = "instance-2";
+
+ CLUSTER.setGroupHeartbeatTimeout(applicationId, heartbeatIntervalMs);
+ CLUSTER.setGroupSessionTimeout(applicationId, sessionTimeoutMs);
+
+ try {
+ try (final KafkaStreams firstStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(firstInstanceId, firstStateDir)
+ );
+ final KafkaStreams secondStreams = new KafkaStreams(
+ setupTopologyWithoutIntermediateUserTopic(),
+ staticStreamsConfig(secondInstanceId, secondStateDir)
+ )) {
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(Arrays.asList(firstStreams, secondStreams));
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasActiveStaticStreamsMember(groupDescription, firstInstanceId) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Streams group did not reach 2 active static members");
+
+ final StreamsGroupDescription initialGroupDescription = describeStreamsGroup(applicationId);
+ final String firstMemberId = staticStreamsMemberId(initialGroupDescription, firstInstanceId).orElseThrow();
+
+ firstStreams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ TestUtils.waitForCondition(() -> {
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 2 &&
+ hasStaticStreamsMemberWithMemberIdAndEpoch(
+ groupDescription,
+ firstInstanceId,
+ firstMemberId,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH
+ ) &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member did not remain in group with static leave epoch");
+
+ final int groupEpochAfterStaticLeave = describeStreamsGroup(applicationId).groupEpoch();
+
+ TestUtils.waitForCondition(() -> {
+ mockTime.sleep(heartbeatIntervalMs);
+
+ final StreamsGroupDescription groupDescription = describeStreamsGroup(applicationId);
+ return groupDescription.members().size() == 1 &&
+ groupDescription.groupEpoch() > groupEpochAfterStaticLeave &&
+ staticStreamsMemberId(groupDescription, firstInstanceId).isEmpty() &&
+ hasActiveStaticStreamsMember(groupDescription, secondInstanceId);
+ }, "Static streams member was not removed after session timeout");
+ }
+ } finally {
+ Utils.delete(firstStateDir);
+ Utils.delete(secondStateDir);
+ }
+ }
+
+ @Test
+ public void testDuplicateStaticInstanceIdDoesNotReplaceExistingStreamsMember() throws Exception {
+ final File firstStateDir = TestUtils.tempDirectory();
Review Comment:
testDuplicateStaticInstanceIdDoesNotReplaceExistingStreamsMember covers the UNRELEASED_INSTANCE_ID path (new join epoch=0 while existing member is active). Worth adding a test for the FENCED_INSTANCE_ID path: stale member with a replaced instanceId reconnects and sends a non-zero epoch heartbeat.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3475788970
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -53,12 +63,7 @@ def __init__(self, test_context):
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
Review Comment:
I meant: Wouldn't the test also work for the streams protocol if we make group.protocol a parameter in the matrix?
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #22561:
URL: https://github.com/apache/kafka/pull/22561
--
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-20169: Add ducktape test code for Streams Rebalance Protocol. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3449382958
##########
tests/kafkatest/tests/streams/streams_static_membership_streams_protocol_test.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
Review Comment:
@lucasbru
Thanks for the review!
You are right. I will fix this.
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3474603672
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -53,12 +63,7 @@ def __init__(self, test_context):
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
Review Comment:
This test still uses the default classic group protocol through `create_processors(self.num_threads)`. I only enable the streams group protocol in the tests that explicitly cover streams-group behavior.
```python
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
...
processors = self.create_processors(self.num_threads)
...
def create_processors(self, num_threads, group_protocol="classic", persistent_process_id_store_enabled=False):
return [ ... ]
```
What do you think?
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4800110628
@lucasbru
Thanks for your time!
Yes, I ran the tests.
Here is test results for 11188ceff2579f73a465d8b90e4112e88aa8daec
```
====================================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.14.0
session_id: 2026-06-25--002
run time: 10 minutes 10.465 seconds
tests run: 5
passed: 5
flaky: 0
failed: 0
ignored: 0
====================================================================================================
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_fencing_static_streams_member.fencing_stage=all.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 1.914 seconds
----------------------------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_fencing_static_streams_member.fencing_stage=stable.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 2 minutes 7.978 seconds
----------------------------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_static_member_process_id_persisted_after_rejoin.bounce_mode=all.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 2 minutes 25.383 seconds
----------------------------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_static_member_process_id_persisted_after_rejoin.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 2 minutes 35.667 seconds
----------------------------------------------------------------------------------------------------
test_id: kafkatest.tests.streams.streams_static_membership_test.StreamsStaticMembershipTest.test_temporary_static_rejoin_does_not_trigger_survivor_reconciliation.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 1 minute 58.915 seconds
----------------------------------------------------------------------------------------------------
```
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4800014673
@chickenchickenlove I'll take a look. Have you run the 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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #22561:
URL: https://github.com/apache/kafka/pull/22561#issuecomment-4831428620
@lucasbru
Good point.
I will create the following PR to address these issues.
Thanks for letting me know 🙇♂️
--
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-20169: Add ducktape test code for static membership in KIP-1071 [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22561:
URL: https://github.com/apache/kafka/pull/22561#discussion_r3473427401
##########
tests/kafkatest/tests/streams/streams_static_membership_test.py:
##########
@@ -53,12 +63,7 @@ def __init__(self, test_context):
def test_rolling_bounces_will_not_trigger_rebalance_under_static_membership(self, metadata_quorum):
Review Comment:
Could we not enable streams groups in this test?
--
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