[kafka] branch trunk updated: MINOR: ignore unused configuration when ConsumerCoordinator is not constructed (#12041)

[guozhang]

This is an automated email from the ASF dual-hosted git repository.

guozhang 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 cf5e714a8b MINOR: ignore unused configuration when ConsumerCoordinator 
is not constructed (#12041)
cf5e714a8b is described below

commit cf5e714a8bea8bb1de75201d0769bb1c246b9334
Author: RivenSun <91005273+rivens...@users.noreply.github.com>
AuthorDate: Fri Apr 15 08:30:43 2022 +0800

    MINOR: ignore unused configuration when ConsumerCoordinator is not 
constructed (#12041)
    
    Following PR #11940, ignore unused config when ConsumerCoordinator is not 
constructed.
    
    Reviewers: Guozhang Wang <wangg...@gmail.com>
---
 .../java/org/apache/kafka/clients/consumer/KafkaConsumer.java    | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git 
a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java 
b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index a49c89560f..6ffb772915 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -774,8 +774,12 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> 
{
             );
 
             // no coordinator will be constructed for the default (null) group 
id
-            this.coordinator = !groupId.isPresent() ? null :
-                new ConsumerCoordinator(groupRebalanceConfig,
+            if (!groupId.isPresent()) {
+                config.ignore(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG);
+                
config.ignore(ConsumerConfig.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
+                this.coordinator = null;
+            } else {
+                this.coordinator = new 
ConsumerCoordinator(groupRebalanceConfig,
                         logContext,
                         this.client,
                         assignors,
@@ -788,6 +792,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
                         
config.getInt(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG),
                         this.interceptors,
                         
config.getBoolean(ConsumerConfig.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED));
+            }
             this.fetcher = new Fetcher<>(
                     logContext,
                     this.client,