由于致命错误，Kafka 消费者轮询方法尝试加入组失败：组成员支持的协议不兼容

Question

我有一个 kafka 消费者处理一个消息主题名为“ABC”的配置：

{key.deserializer=class org.apache.kafka.common.serialization.StringDeserializer, value.deserializer=class org.apache.kafka.common.serialization.StringDeserializer, max.poll.records=250, group.instance。 id=ABC, group.id=ABC, bootstrap.servers=localhost:9093, auto.commit.interval.ms=50, security.protocol=SSL, enable.auto.commit=true, ssl.truststore.location=, ssl.truststore.password=, ssl.endpoint.identification.algorithm=, client.id=ABC}

消费者代码是：

            Runnable processMessageConsumer = () -> {

                try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig)) {

                    log.debug("consumer created {} about to subscribe to topic ABC", consumer);
                    consumer.subscribe(Collections.singletonList("ABC"));
                    log.debug("subscribed to topic ABC {}", consumer);

                    while(!Thread.currentThread().isInterrupted()) {
                        try {
                            log.debug("about to call consumer.poll");
                            final ConsumerRecords<String, String> records = consumer.poll(Duration.ofMinutes(1));

                            log.debug("return from consumer.poll got records of {}", records);

                            if (records.isEmpty()) {
                                log.warn("Message receive queue is silent (no message received over kafka).");
                            } else {
                                log.debug("about to call processReceivedMessage");
                                processReceivedMessage(records);
                                consumer.commitAsync();
                            }
                        } catch (Exception ex) {
                            log.error("Failed to process received message", ex);
                        }

                    }
            };

maven pom kafka 依赖：

</properties>
 <kafka.version>2.3.0</kafka.version>
...
</properties> 
...
<dependencies>
       <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>${kafka.version}</version>
        </dependency>

</dependencies>

该代码直到最近一直运行良好，但现在出现此错误：

- DEBUG [ MSG-Consumer-0] n.r.r.s.i.KafkaService$$EnhancerBySpringCGLIB$$30ee2b7b                         :367  about to call consumer.poll
- ERROR [ MSG-Consumer-0] o.a.k.c.c.internals.AbstractCoordinator                                          :569  [Consumer clientId=ABC, groupId=ABC] Attempt to join group failed due to fatal error: The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list.
- ERROR [ MSG-Consumer-0] n.r.r.s.i.KafkaService$$EnhancerBySpringCGLIB$$30ee2b7b                          :381  Failed to process received message
-
org.apache.kafka.common.errors.InconsistentGroupProtocolException: The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list.

Answer 1

所以我现在已经弄清楚了问题所在。基本上，我们的库中有标准的 kakfa 代码来执行通用消息传递，其中每个服务都有自己的“收件箱”主题，它从其他服务接收消息。主题以服务命名，在这种情况下，ABC 是服务。因此，如果 XYZ 服务想要向 ABC 服务发送消息，它会将消息放在 ABC 主题上。消费者线程然后计算出如何将特定类型的消息分派到相关端点（消息处理方法）。而且...为了方便起见，消费者还将 group.id 设置为 ABC。

这种机制一直很好地工作，直到最近一些流消费者代码也被添加到库中，该库使用自己的特定主题，但将 application.id（用于流配置）再次设置为服务名称在这种情况下将是“ABC”。这就是导致 InconsistentGroupProtocolException 的原因。

我相信在幕后，application.id 被用作 group.id，所以我们有不同的主题，但 group.id 相同。而卡夫卡似乎不太喜欢这个！修复很简单，我只是将流处理程序的类名添加到 application.id 并解决了它。

所以对于不同的主题使用不同的 group.id/application.id