KAFKA-17338 ConsumerConfig should prevent using partition assignors with CONSUMER group protocol #16899

m1a2st · 2024-08-16T13:59:06Z

Jira: https://issues.apache.org/jira/browse/KAFKA-17338
add a validator for consumer group protocol, when setting partition.assignment.strategy

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712

@m1a2st thanks for this patch

chia7712 · 2024-08-22T06:48:56Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+    private void checkPartitionAssigmentStrategy() {
+        List<String> assignmentStrategies = getList(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);
+        if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {
+            List<Class<? extends AbstractPartitionAssignor>> defaultValue = Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class);


Could you please create a default value of Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class)? That can keep the consistency in the futrue

chia7712 · 2024-08-22T06:51:48Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+        List<String> assignmentStrategies = getList(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);
+        if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {
+            List<Class<? extends AbstractPartitionAssignor>> defaultValue = Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class);
+            if (!new HashSet<>(assignmentStrategies).containsAll(defaultValue) || assignmentStrategies.size() != defaultValue.size()) {


maybe we should accept the config equal to the default. for example, the config having org.apache.kafka.clients.consumer.RangeAssignor and org.apache.kafka.clients.consumer.CooperativeStickyAssignor should work

m1a2st · 2024-08-22T12:06:18Z

@chia7712, Thanks for your comments, I have addressed all the comments, PTAL

chia7712 · 2024-08-31T11:36:18Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+                .map(Class::getName).collect(Collectors.toSet());
+        if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {
+            if ((!new HashSet<>(assignmentStrategies).containsAll(defaultAssignmentStrategiesClassNames) &&
+                    !new HashSet<>(assignmentStrategies).containsAll(PARTITION_ASSIGNOR_DEFAULT_VALUE)) ||


how about:

@SuppressWarnings("unchecked") private void checkPartitionAssigmentStrategy() { List<Object> assignmentStrategies = (List<Object>) get(PARTITION_ASSIGNMENT_STRATEGY_CONFIG); Set<String> defaultAssignmentStrategiesClassNames = PARTITION_ASSIGNOR_DEFAULT_VALUE.stream() .map(Class::getName).collect(Collectors.toSet()); if (PARTITION_ASSIGNOR_DEFAULT_VALUE.size() != assignmentStrategies.size() || !PARTITION_ASSIGNOR_DEFAULT_VALUE.stream().allMatch(clz -> assignmentStrategies.stream().anyMatch(obj -> clz.equals(obj) || clz.getName().equals(obj)))) { throw new ConfigException(PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " cannot be set when " + GROUP_PROTOCOL_CONFIG + "=" + GroupProtocol.CONSUMER.name()); } }

It is great, I will address it

chia7712 · 2024-09-01T03:02:17Z

@m1a2st please fix the failed tests

chia7712 · 2024-09-01T16:16:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -384,7 +385,9 @@ public class ConsumerConfig extends AbstractConfig {
    private static final String SECURITY_PROVIDERS_DOC = SecurityConfig.SECURITY_PROVIDERS_DOC;

    private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
-
+    private static final List<Class<? extends AbstractPartitionAssignor>> PARTITION_ASSIGNOR_DEFAULT_VALUE = 
+            Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class);


Please make it immutable

chia7712 · 2024-09-03T17:52:18Z

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java

+        final Map<String, Object> configs = new HashMap<>();
+        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass);
+        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass);
+        configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RangeAssignor,org.apache.kafka.clients.consumer.CooperativeStickyAssignor");


Could you please add following test cases?

@Test public void testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol() { testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol("org.apache.kafka.clients.consumer.RangeAssignor,org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol("org.apache.kafka.clients.consumer.CooperativeStickyAssignor, org.apache.kafka.clients.consumer.RangeAssignor"); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList("org.apache.kafka.clients.consumer.RangeAssignor", CooperativeStickyAssignor.class)); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(CooperativeStickyAssignor.class, "org.apache.kafka.clients.consumer.RangeAssignor")); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(RangeAssignor.class, "org.apache.kafka.clients.consumer.CooperativeStickyAssignor")); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList("org.apache.kafka.clients.consumer.CooperativeStickyAssignor", RangeAssignor.class)); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(CooperativeStickyAssignor.class, RangeAssignor.class)); testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class)); } private void testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Object value) { final Map<String, Object> configs = new HashMap<>(); configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass); configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass); configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, value); configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name()); assertDoesNotThrow(() -> new ConsumerConfig(configs)); }

kirktrue

Thanks for the PR @m1a2st!

I'd like to see if we can simplify this change. It seems by using originals(), we can clean up the checkPartitionAssignmentStrategy() method, which then allows us to revert the changes that introduced PARTITION_ASSIGNOR_DEFAULT_VALUE.

LMK if I'm missing something, though.

Thanks!

kirktrue · 2024-09-03T20:34:31Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+    @SuppressWarnings("unchecked")
+    private void checkPartitionAssigmentStrategy() {
+        List<Object> assignmentStrategies = (List<Object>) get(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);
+        if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {
+            if (PARTITION_ASSIGNOR_DEFAULT_VALUE.size() != assignmentStrategies.size() ||
+                    !PARTITION_ASSIGNOR_DEFAULT_VALUE.stream()
+                            .allMatch(clz -> assignmentStrategies.stream()
+                                    .anyMatch(obj -> clz.equals(obj) || clz.getName().equals(obj)))
+            ) {
+                throw new ConfigException(PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " cannot be set when " + GROUP_PROTOCOL_CONFIG + "=" + GroupProtocol.CONSUMER.name());
+            }
+        }
+    }
+


I think the core check could be much simpler 🤔

I used the originals() method to check the values the user provided before any defaults were applied.

Suggested change

@SuppressWarnings("unchecked")

private void checkPartitionAssigmentStrategy() {

List<Object> assignmentStrategies = (List<Object>) get(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);

if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {

if (PARTITION_ASSIGNOR_DEFAULT_VALUE.size() != assignmentStrategies.size() ||

!PARTITION_ASSIGNOR_DEFAULT_VALUE.stream()

.allMatch(clz -> assignmentStrategies.stream()

.anyMatch(obj -> clz.equals(obj) || clz.getName().equals(obj)))

) {

throw new ConfigException(PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " cannot be set when " + GROUP_PROTOCOL_CONFIG + "=" + GroupProtocol.CONSUMER.name());

}

}

}

private void checkPartitionAssignmentStrategy() {

if (!getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name()))

return;

Object config = originals().get(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);

if (config != null && !Utils.isBlank(config.toString())) {

throw new ConfigException(PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " cannot be set when " + GROUP_PROTOCOL_CONFIG + "=" + GroupProtocol.CONSUMER.name());

}

}

Is there anything missing in the above?

Hello @kirktrue, Thanks for your comments, I think we have two scenario to discuss this issue
If we want to block any input of partition.assignment.strategy, your method is more simplify and better, however if we allow user to input the default value of partition.assignment.strategy, we use the origin method is more easier to check.

I would say we want to throw if there is any value defined by the user for this property (it's truly deprecated with the new protocol), so @kirktrue 's suggestion of working with the originals seems effective and simpler.

+1 to @kirktrue @lianetm that checking the user-defined values makes sense to me.

For another, ConsumerConfig's constructor will log all configs, so it would be nice to set the config partition.assignment.strategy to null when the protocol is CONSUMER. This behavior is similar to maybeOverrideEnableAutoCommit which can avoid printing "weird" consumer configs in log file.

Do you refer to the log for unused? I did think about that, but then checked and noticed that it only logs based on the originals, not the defaults , so seems we shouldn't need to override to avoid the log

kafka/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

Line 225 in 005155b

Set<String> keys = new HashSet<>(originals.keySet());

Do you refer to the log for unused?

I think about logAll()

kafka/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

Line 121 in 005155b

logAll();

If we don't remove the default value of partition.assignment.strategy, it will print following log.

[2024-09-04 19:45:40,533] INFO ConsumerConfig values: ... group.protocol = CONSUMER ... partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]

That is not a big issue, but the output could misdirect users to assume "those" configs are valid.

It make sense to me. +1 to @chia7712, the config log should not misdirect user.

totally agree it will be confusing on the logAll, +1 to remove

@m1a2st I think this comment can be addressed in a follow-up (https://issues.apache.org/jira/browse/KAFKA-17821). Could you please revert the related changes? Sorry for complicating this PR.

lianetm · 2024-09-04T11:13:37Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+    @SuppressWarnings("unchecked")
+    private void checkPartitionAssigmentStrategy() {
+        List<Object> assignmentStrategies = (List<Object>) get(PARTITION_ASSIGNMENT_STRATEGY_CONFIG);
+        if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {


I wonder if it would help readability to encapsulate this in something along the lines of:

Suggested change

if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {

if (isConsumerProtocol()) {

And then a similar isClassicProtocol(). The trick is that we already have protocol checks in 2 places (this and checkGroupRemoteAssignor), and I expect we will need it again as we keep adding other consistency validations.

It's a good idea, I will extract these method.

lianetm · 2024-09-04T13:43:42Z

High level comment, there are other properties in the exact same situation, basically properties that do not make sense with the new protocol, and we want to throw ConfigException if the user provides a value for them (we also want to remove the defaults as @chia7712 pointed out, to remove them from the log and avoid confusion).

Ok with me if we prefer to tackle them separately (I can file the jiras), but the implementation of the check seems the same, so wonder if we should consider it here, and instead of having a checkPartitionAssigmentStrategy (that we'll end replicating for each of these props), we have a single checkDeprecatedPropertyUsed(propId) (or a better name), doing the exact same check that we're considering here for the partition strategy. I expect we would need the same check for all these that are deprecated with the new protocol (still thinking if I'm missing another one):

heartbeat.interval.ms
session.timeout.ms
group.max.session.timeout.ms
group.mix.session.timeout.ms

Thoughts?

m1a2st · 2024-09-04T15:01:15Z

I think that I can address this check method in this PR, and remove these config default value when user config the consumer group protocol.

chia7712 · 2024-09-04T15:12:38Z

heartbeat.interval.ms
session.timeout.ms
group.max.session.timeout.ms
group.mix.session.timeout.ms

+1 and thanks @m1a2st to address them in this PR :)

chia7712 · 2024-09-05T13:57:19Z

group.max.session.timeout.ms
group.mix.session.timeout.ms

Those configs are used by server rather than client. The new coordinator supports CLASSIC protocol so we don't need to unset those configs I think.

kirktrue · 2024-10-08T22:57:22Z

@m1a2st—I wanted to make sure that you weren't held up waiting for input from others. After the number of review comments gets to a certain size, it can be hard to tell where we stand. Let me know if you need us to review or respond to any outstanding questions. Thanks!

m1a2st · 2024-10-09T14:48:19Z

@kirktrue,
Thank you for your comments. I have a question regarding this:

Should we treat this as an issue and resolve it?
Or, is it not considered an issue, and setting it to zero is the appropriate solution?

chia7712 · 2024-10-17T13:18:56Z

@m1a2st could you please rebase code?

chia7712 · 2024-10-26T22:09:29Z

@m1a2st please check the failed tests

kirktrue

Thanks for the changes @m1a2st. Just a few more comments, mostly formatting.

kirktrue · 2024-10-30T21:45:58Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+    /**
+     * A list of configuration keys for CLASSIC protocol not supported. we should check the input string and clean up the
+     * default value.
+     */
+    private static final List<String> CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS = Collections.singletonList(
+            GROUP_REMOTE_ASSIGNOR_CONFIG
+    );
+
+    /**
+     * A list of configuration keys for consumer protocol not supported. we should check the input string and clean up the
+     * default value.
+     */
+    private static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = 
+            List.of(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, HEARTBEAT_INTERVAL_MS_CONFIG, SESSION_TIMEOUT_MS_CONFIG);
+


Can we change this to a Map<GroupProtocol, Set<String>> GROUP_PROTOCOL_UNSUPPORTED_CONFIGS to make it a little more concise?

kirktrue · 2024-10-30T21:47:31Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

-        Collections.unmodifiableList(Arrays.asList(
-            RANGE_ASSIGNOR_NAME,
-            ROUNDROBIN_ASSIGNOR_NAME,
-            STICKY_ASSIGNOR_NAME,
-            COOPERATIVE_STICKY_ASSIGNOR_NAME
-        ));
+            List.of(RANGE_ASSIGNOR_NAME, ROUNDROBIN_ASSIGNOR_NAME, STICKY_ASSIGNOR_NAME, COOPERATIVE_STICKY_ASSIGNOR_NAME);


Can we keep the configuration on each line like it is now? That allows the list to expand or contract without messing with line wrapping. But this is good:

- Collections.unmodifiableList(Arrays.asList( + List.of(

kirktrue · 2024-10-30T21:57:25Z

core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala

@@ -83,8 +83,10 @@ abstract class BaseConsumerTest extends AbstractConsumerTest {
  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
  def testCoordinatorFailover(quorum: String, groupProtocol: String): Unit = {
    val listener = new TestConsumerReassignmentListener()
-    this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "5001")
-    this.consumerConfig.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "1000")
+    if (groupProtocol.contains("CONSUMER")) {


Why do we only set these for the new consumer?

This is a mistake, fix it

kirktrue · 2024-10-30T21:59:52Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+     * default value.
+     */
+    private static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = 
+            List.of(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, HEARTBEAT_INTERVAL_MS_CONFIG, SESSION_TIMEOUT_MS_CONFIG);


@lianetm earlier mentioned these configuration as being invalid for the new consumer:

heartbeat.interval.ms

session.timeout.ms

group.max.session.timeout.ms

group.mix.session.timeout.ms

It looks like we have the first two. Did we need to add the other two group session timeout configuration to that list?

Yes you are right, If this PR didn't need to deal with log problem, we should add group.max.session.timeout.ms and group.mix.session.timeout.ms to CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS list.

m1a2st · 2024-10-31T10:35:27Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+            "group.max.session.timeout.ms",
+            "group.mix.session.timeout.ms"


If use GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG it will have a circle dependency, thus I use string, Im not sure it a good way to do it, or we should check it at `GroupCoordinatorConfig.

well these configs are used by the broker only, not the client, so I would say we don't deal with them here and let the broker drive it?

m1a2st · 2024-10-31T10:40:28Z

core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala

@@ -83,8 +83,10 @@ abstract class BaseConsumerTest extends AbstractConsumerTest {
  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
  def testCoordinatorFailover(quorum: String, groupProtocol: String): Unit = {
    val listener = new TestConsumerReassignmentListener()
-    this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "5001")
-    this.consumerConfig.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "1000")
+    if (groupProtocol.contains("CONSUMER")) {


This is a mistake, fix it

kirktrue · 2024-11-20T01:14:29Z

@m1a2st—just checking in on this PR. It seems like we're really close. Is there anything you need from @lianetm or myself? Thanks!

# Conflicts: # tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java # tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java

m1a2st · 2024-11-20T10:52:03Z

Hello @kirktrue, I think if this PR pass the CI, and won't have comments, we should merge this PR, WDYT?

lianetm

Thanks for this patch @m1a2st, just some minor comments for consideration.

lianetm · 2024-11-21T21:03:25Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java


+    /**
+     * A list of configuration keys for CLASSIC protocol not supported. we should check the input string and clean up the


Suggested change

* A list of configuration keys for CLASSIC protocol not supported. we should check the input string and clean up the

* A list of configuration keys not supported for CLASSIC protocol.

I'm also suggesting we remove the second sentence about "clean up default values" because I cannot find that we clean up values for unsupported configs, I only see we throw ConfigException. Am I missing something?

lianetm · 2024-11-21T21:08:50Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+    );
+
+    /**
+     * A list of configuration keys for consumer protocol not supported. we should check the input string and clean up the


Suggested change

* A list of configuration keys for consumer protocol not supported. we should check the input string and clean up the

* A list of configuration keys not supported for CONSUMER protocol.

lianetm · 2024-11-21T21:18:00Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+            "group.max.session.timeout.ms",
+            "group.mix.session.timeout.ms"


well these configs are used by the broker only, not the client, so I would say we don't deal with them here and let the broker drive it?

lianetm · 2024-11-21T21:25:53Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

+            unsupportedConfigs.forEach(configName -> {
+                Object config = originals().get(configName);
+                if (config != null && !Utils.isBlank(config.toString())) {
+                    throw new ConfigException(configName + " cannot be set when " + GROUP_PROTOCOL_CONFIG + "=" + groupProtocol.name());


would it be a better experience for the user if we gather all the unsupported configs used first, and then throw an error with all of them? I expect it will be helpful when users upgrade to the new protocol, and probably leave the existing config (with several unsupported).

lianetm · 2024-11-21T21:27:29Z

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java

+        final Map<String, Object> configs = new HashMap<>();
+        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializerClass);
+        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializerClass);
+        configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name());
+        configs.put(configName, value);


lianetm · 2024-11-21T21:29:24Z

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java

+        configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name());
+        configs.put(configName, value);
+        ConfigException exception = assertThrows(ConfigException.class, () -> new ConsumerConfig(configs));
+        assertTrue(exception.getMessage().contains(configName + 


can't we assertEquals here ?

lianetm · 2024-11-21T22:04:08Z

tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java

@@ -337,7 +335,6 @@ private Map<String, Object> composeConfigs(ClusterInstance cluster, String group
        configs.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(GROUP_PROTOCOL_CONFIG, groupProtocol);
-        configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());


just to double check, even though we're removing this prop, I expect the test will still use the range assignor if under classic protocol just because range is the first assignor in the default value of the config. Correct?

lianetm · 2024-11-21T22:19:15Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -413,7 +434,7 @@ public class ConsumerConfig extends AbstractConfig {
                                        HEARTBEAT_INTERVAL_MS_DOC)
                                .define(PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
                                        Type.LIST,
-                                        Arrays.asList(RangeAssignor.class, CooperativeStickyAssignor.class),
+                                        PARTITION_ASSIGNOR_DEFAULT_VALUE,


This constant was introduced to be reused and that made sense, but seems it's only used here now right? If so I think keeping the default explicitly here as it was is more convenient, it just makes it easier to discover (I personally navigate to this definition a lot looking for defaults :))

# Conflicts: # clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

m1a2st · 2024-11-23T03:07:31Z

Thanks for @lianetm review, addressed all comments

m1a2st changed the title ~~KAFKA-17338~~ KAFKA-17338 ConsumerConfig should prevent using partition assignors with CONSUMER group protocol Aug 17, 2024

m1a2st marked this pull request as ready for review August 20, 2024 08:02

chia7712 reviewed Aug 22, 2024

View reviewed changes

chia7712 reviewed Aug 31, 2024

View reviewed changes

chia7712 reviewed Sep 1, 2024

View reviewed changes

chia7712 reviewed Sep 3, 2024

View reviewed changes

kirktrue suggested changes Sep 3, 2024

View reviewed changes

kirktrue added consumer KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels Sep 3, 2024

lianetm reviewed Sep 4, 2024

View reviewed changes

m1a2st added 10 commits September 4, 2024 20:22

validate PARTITION_ASSIGNMENT_STRATEGY_CONFIG in consumer protocol

5f17f52

fix checkPartitionAssigmentStrategy logic problem

38fa6c9

fix default value logic

59da76d

refactor checkPartitionAssigmentStrategy method

2ee47e6

fix test

e8354a6

change PARTITION_ASSIGNOR_DEFAULT_VALUE into immutable list

ec363a1

refactor checkPartitionAssignmentStrategy method

9117089

remove unused test

be90c63

remove unused test

6c975f5

use spotless

8bcbdee

m1a2st force-pushed the KAFKA-17338 branch from f7a54db to 8bcbdee Compare September 4, 2024 12:22

m1a2st added 2 commits September 5, 2024 20:15

add check for consumer protocol

8846272

update test

0e77b0d

kirktrue added the clients label Sep 26, 2024

Merge branch 'trunk' into KAFKA-17338

a03d614

github-actions bot added tools small Small PRs labels Oct 17, 2024

revert print log issue code

07a7a7e

m1a2st force-pushed the KAFKA-17338 branch from 7bd6b42 to 07a7a7e Compare October 17, 2024 14:25

fix test error

b81c90a

github-actions bot added the core Kafka Broker label Oct 27, 2024

m1a2st added 2 commits October 27, 2024 12:43

fix test error

baa285e

fix test error

c306364

github-actions bot removed the small Small PRs label Oct 27, 2024

kirktrue suggested changes Oct 30, 2024

View reviewed changes

kirktrue reviewed Oct 30, 2024

View reviewed changes

m1a2st added 2 commits October 31, 2024 18:32

addressed by comments

0841459

addressed by comments

76e5c85

m1a2st commented Oct 31, 2024

View reviewed changes

m1a2st added 2 commits November 20, 2024 18:48

Merge branch 'trunk' into KAFKA-17338

15d3255

# Conflicts: # tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java # tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java

remove unused import

8fc8205

lianetm reviewed Nov 21, 2024

View reviewed changes

m1a2st added 2 commits November 23, 2024 09:32

Merge branch 'trunk' into KAFKA-17338

fbf606b

# Conflicts: # clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

addressed by comments

c0bb015

m1a2st commented Nov 23, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java Outdated Show resolved Hide resolved

addressed by comments

abfb729

	if (getString(GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.CONSUMER.name())) {
	if (isConsumerProtocol()) {

		"group.max.session.timeout.ms",
		"group.mix.session.timeout.ms"


		/**
		* A list of configuration keys for CLASSIC protocol not supported. we should check the input string and clean up the

	* A list of configuration keys for CLASSIC protocol not supported. we should check the input string and clean up the
	* A list of configuration keys not supported for CLASSIC protocol.

	* A list of configuration keys for consumer protocol not supported. we should check the input string and clean up the
	* A list of configuration keys not supported for CONSUMER protocol.

KAFKA-17338 ConsumerConfig should prevent using partition assignors with CONSUMER group protocol #16899

Are you sure you want to change the base?

KAFKA-17338 ConsumerConfig should prevent using partition assignors with CONSUMER group protocol #16899

Conversation

m1a2st commented Aug 16, 2024 • edited Loading

Committer Checklist (excluded from commit message)

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m1a2st commented Aug 22, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Sep 1, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m1a2st Sep 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m1a2st Sep 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm commented Sep 4, 2024 • edited Loading

m1a2st commented Sep 4, 2024

chia7712 commented Sep 4, 2024

chia7712 commented Sep 5, 2024

kirktrue commented Oct 8, 2024

m1a2st commented Oct 9, 2024

chia7712 commented Oct 17, 2024

chia7712 commented Oct 26, 2024

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirktrue commented Nov 20, 2024

m1a2st commented Nov 20, 2024

lianetm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m1a2st commented Nov 23, 2024

m1a2st commented Aug 16, 2024 •

edited

Loading

m1a2st Sep 4, 2024 •

edited

Loading

m1a2st Sep 4, 2024 •

edited

Loading

lianetm commented Sep 4, 2024 •

edited

Loading