KAFKA-19376: Throw an error message if any unsupported feature is used with KIP-1071 #19908

lucasbru · 2025-06-05T13:59:25Z

We should be mindful of ours users and let them know early if they are
using an unsupported feature in 4.1.

Unsupported features:

Regular expressions
Warm-up replicas (high availability assignor)
Static membership
Standby replicas enabled through local config
Named topologies (already checked)
Non-default kafka-client supplier

lucasbru · 2025-06-05T14:01:13Z

Copilot

Pull Request Overview

This PR introduces additional validations for the STREAMS protocol by throwing errors or logging warnings when unsupported configuration features are used.

Tests are added to verify that unsupported features (pattern subscriptions, warmup replicas, static membership, non-default client supplier) produce the correct error messages or warnings.
Internal code in StreamsConfig and KafkaStreams now enforce these validations, and a redundant log message is removed from StreamThread.

Reviewed Changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java	Added tests to validate warning logs and exception throwing for unsupported features with STREAMS protocol.
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java	Added tests to ensure that using pattern subscriptions and non-default client supplier with STREAMS protocol throws exceptions.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java	Removed redundant logging of STREAMS protocol activation.
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java	Implemented compatibility verification for static membership, warmup replicas, and standby replicas when using STREAMS protocol.
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java	Introduced additional unsupported feature checks and error throwing for named topologies, pattern subscriptions, and non-default kafka client supplier.

Copilot · 2025-06-05T14:01:32Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -1047,6 +1051,21 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
        rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, applicationConfigs);
    }

+    private void throwIfUnsupportedFeatureIsUsedWithStreamsRebalanceProtocol() {
+        if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {


Consider invoking GroupProtocol.STREAMS.name() (with parentheses) instead of GroupProtocol.STREAMS.name to correctly compare the protocol string.

Suggested change

if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {

if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name())) {

bbejeck

Thanks for the PR @lucasbru - overall this LGTM but the changes here cause the KafkaStreamsTelemetryIntegrationTest test to fail as it uses its own client supplier to get access to the passed metrics.

chia7712

@lucasbru thanks for this patch. Two small questions remain. PTAL

chia7712 · 2025-06-05T16:26:20Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java


+    private void verifyStreamsProtocolCompatibility(final boolean doLog) {
        if (doLog && getString(GROUP_PROTOCOL_CONFIG).equals(GroupProtocol.STREAMS.name().toLowerCase(Locale.ROOT))) {


Should equals be replaced by equalsIgnoreCase, since the other checks, such as throwIfUnsupportedFeatureIsUsedWithStreamsRebalanceProtocol and checkUnsupportedConfigsPostProcess, use equalsIgnoreCase?

Good point. I actually added a little protected method to check if streams protocol is enabled, to eliminate this source of bugs.

chia7712 · 2025-06-05T16:52:35Z

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
+        props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");
+        props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "static-member-1");


should we test the static id with consumer prefix?

Good point. Updated

lucasbru

@chia7712 @bbejeck Thanks for the comments! All addressed, ready for re-review.

For the Kafka Client supplier, I demoted the check to a warning instead of throwing an error, since it is still useful to supply a custom producer or admin client. This also fixes the failing integration test.

lucasbru · 2025-06-06T09:33:25Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -1047,6 +1051,21 @@ private KafkaStreams(final TopologyMetadata topologyMetadata,
        rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, applicationConfigs);
    }

+    private void throwIfUnsupportedFeatureIsUsedWithStreamsRebalanceProtocol() {
+        if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {


lucasbru · 2025-06-06T09:34:26Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java


+    private void verifyStreamsProtocolCompatibility(final boolean doLog) {
        if (doLog && getString(GROUP_PROTOCOL_CONFIG).equals(GroupProtocol.STREAMS.name().toLowerCase(Locale.ROOT))) {


Good point. I actually added a little protected method to check if streams protocol is enabled, to eliminate this source of bugs.

lucasbru · 2025-06-06T09:40:31Z

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
+        props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");
+        props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "static-member-1");


Good point. Updated

impl

ff3149b

github-actions bot added the streams label Jun 5, 2025

lucasbru requested review from bbejeck, cadonna, Copilot and mjsax June 5, 2025 14:00

Copilot AI reviewed Jun 5, 2025

View reviewed changes

bbejeck reviewed Jun 5, 2025

View reviewed changes

chia7712 reviewed Jun 5, 2025

View reviewed changes

address comments

776d8bb

lucasbru commented Jun 6, 2025

View reviewed changes

lucasbru added 2 commits June 6, 2025 11:43

checkstyle

c2e9e40

test fix

56b2cb3

chia7712 approved these changes Jun 7, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

KAFKA-19376: Throw an error message if any unsupported feature is used with KIP-1071 #19908

KAFKA-19376: Throw an error message if any unsupported feature is used with KIP-1071 #19908

Uh oh!

lucasbru commented Jun 5, 2025 •

edited by github-actions bot

Loading

Uh oh!

lucasbru commented Jun 5, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Jun 5, 2025

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

bbejeck left a comment

Uh oh!

chia7712 left a comment

Uh oh!

chia7712 Jun 5, 2025

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

chia7712 Jun 5, 2025

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

lucasbru left a comment

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

lucasbru Jun 6, 2025

Uh oh!

Uh oh!

	if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {
	if (applicationConfigs.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name())) {


		private void verifyStreamsProtocolCompatibility(final boolean doLog) {
		if (doLog && getString(GROUP_PROTOCOL_CONFIG).equals(GroupProtocol.STREAMS.name().toLowerCase(Locale.ROOT))) {

KAFKA-19376: Throw an error message if any unsupported feature is used with KIP-1071 #19908

Are you sure you want to change the base?

KAFKA-19376: Throw an error message if any unsupported feature is used with KIP-1071 #19908

Uh oh!

Conversation

lucasbru commented Jun 5, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lucasbru commented Jun 5, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lucasbru commented Jun 5, 2025 •

edited by github-actions bot

Loading