You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/08/26 12:02:02 UTC
(kafka) branch trunk updated: KAFKA-17324: Set config group.protocol to classic in Streams (#16878)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 4886414aa0e KAFKA-17324: Set config group.protocol to classic in Streams (#16878)
4886414aa0e is described below
commit 4886414aa0edcca67ae69399a34cd2bc9c225239
Author: TengYao Chi <ki...@gmail.com>
AuthorDate: Mon Aug 26 20:01:56 2024 +0800
KAFKA-17324: Set config group.protocol to classic in Streams (#16878)
Streams is not compatible with the new consumer rebalance protocol proposed in KIP-848. Thus, Streams should set/override config group.protocol to classic at startup to ensure that the classic protocol is used.
Reviewers: Lucas Brutschy <lb...@confluent.io>
---
.../org/apache/kafka/streams/StreamsConfig.java | 3 ++-
.../apache/kafka/streams/StreamsConfigTest.java | 24 ++++++++++++++++++++++
2 files changed, 26 insertions(+), 1 deletion(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index b232d76981a..e0c39e105ae 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -823,7 +823,7 @@ public class StreamsConfig extends AbstractConfig {
private static final String[] NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS =
- new String[] {ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG};
+ new String[] {ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.GROUP_PROTOCOL_CONFIG};
private static final String[] NON_CONFIGURABLE_CONSUMER_EOS_CONFIGS =
new String[] {ConsumerConfig.ISOLATION_LEVEL_CONFIG};
private static final String[] NON_CONFIGURABLE_PRODUCER_EOS_CONFIGS =
@@ -1225,6 +1225,7 @@ public class StreamsConfig extends AbstractConfig {
tempConsumerDefaultOverrides.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
tempConsumerDefaultOverrides.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
tempConsumerDefaultOverrides.put("internal.leave.group.on.close", false);
+ tempConsumerDefaultOverrides.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "classic");
CONSUMER_DEFAULT_OVERRIDES = Collections.unmodifiableMap(tempConsumerDefaultOverrides);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index 8e936f328ea..eee52427723 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -470,6 +470,22 @@ public class StreamsConfigTest {
assertEquals("false", consumerConfigs.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
}
+ @Test
+ public void shouldResetToDefaultIfConsumerGroupProtocolIsOverridden() {
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.GROUP_PROTOCOL_CONFIG), "consumer");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs("a", "b", threadIdx);
+ assertEquals("classic", consumerConfigs.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+ }
+
+ @Test
+ public void shouldResetToDefaultIfRestoreConsumerGroupProtocolIsOverridden() {
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.GROUP_PROTOCOL_CONFIG), "consumer");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ final Map<String, Object> consumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
+ assertEquals("classic", consumerConfigs.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+ }
+
@Test
public void testGetRestoreConsumerConfigsWithRestoreConsumerOverriddenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");
@@ -518,6 +534,14 @@ public class StreamsConfigTest {
assertEquals("false", consumerConfigs.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
}
+ @Test
+ public void shouldResetToDefaultIfGlobalConsumerGroupProtocolIsOverridden() {
+ props.put(StreamsConfig.consumerPrefix(ConsumerConfig.GROUP_PROTOCOL_CONFIG), "consumer");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
+ final Map<String, Object> consumerConfigs = streamsConfig.getGlobalConsumerConfigs(clientId);
+ assertEquals("classic", consumerConfigs.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG));
+ }
+
@Test
public void testGetGlobalConsumerConfigsWithGlobalConsumerOverriddenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");