如何为Kafka Streams偏移量提交失败错误实现异常处理程序？

应用环境 #

• Java 17
• Kafka Streams 3.4.0

问题场景 #

日志中出现如下错误：

2025-02-02 20:35:15.685 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler [ERROR] - [Consumer clientId=topic-cg-002417fe-5ce2-4bbc-ad2a-ed3ac0a4901c-StreamThread-4-consumer, groupId=topic-cg] Offset commit failed on partition topic-21 at offset 409115: The coordinator is not aware of this member.

需要捕获这类错误，把核心异常消息Offset commit failed on partition topic-21 at offset 409115: The coordinator is not aware of this member上报到Dynatrace（上报逻辑已实现，核心要解决的是怎么写这个错误处理程序）。

最佳实现方案 #

这类错误是Kafka Consumer内部线程抛出的，常规的Kafka Streams错误处理器（比如StreamsUncaughtExceptionHandler）抓不到，这里给两种靠谱的实现方式：

方案1：自定义日志拦截器（优先推荐） #

直接通过日志框架拦截指定类的ERROR日志，精准匹配目标错误：

1. 基于你用的日志框架（比如Logback）写一个自定义Appender，监听org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler类的ERROR级日志
2. 过滤出包含Offset commit failed和The coordinator is not aware of this member的日志消息
3. 提取消息内容，调用已有的Dynatrace上报逻辑

示例代码（Logback）：

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;

public class DynatraceOffsetErrorAppender extends AppenderBase<ILoggingEvent> {
    @Override
    protected void append(ILoggingEvent event) {
        // 过滤指定日志类和级别
        if ("org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler".equals(event.getLoggerName())
            && event.getLevel().isGreaterOrEqual(ch.qos.logback.classic.Level.ERROR)) {
            String logMsg = event.getFormattedMessage();
            // 匹配目标错误模式
            if (logMsg.contains("Offset commit failed") && logMsg.contains("The coordinator is not aware of this member")) {
                // 调用已实现的Dynatrace上报方法
                DynatraceReporter.reportError(logMsg);
            }
        }
    }
}

Logback配置文件添加Appender：

<appender name="DYNATRACE_OFFSET_ERROR" class="com.yourpackage.DynatraceOffsetErrorAppender" />
<logger name="org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler" level="ERROR">
    <appender-ref ref="DYNATRACE_OFFSET_ERROR" />
</logger>

方案2：自定义Consumer拦截器 #

通过实现ConsumerInterceptor，在偏移提交环节捕获异常：

1. 实现ConsumerInterceptor接口，重写onCommit方法
2. 捕获UnknownMemberIdException（对应错误提示"The coordinator is not aware of this member"）
3. 构造错误消息并上报到Dynatrace

示例代码：

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnknownMemberIdException;

import java.util.Map;

public class OffsetCommitErrorInterceptor implements ConsumerInterceptor<Object, Object> {
    @Override
    public void configure(Map<String, ?> configs) {}

    @Override
    public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> records) {
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // Kafka内部提交失败时会抛出异常，这里捕获对应类型
        try {
            // 空实现，异常由上层调用抛出
        } catch (UnknownMemberIdException e) {
            // 构造错误消息，可结合上下文补充分区、偏移量细节
            String errorMsg = String.format("Offset commit failed: %s", e.getMessage());
            DynatraceReporter.reportError(errorMsg);
        } catch (Exception e) {
            // 其他异常可按需处理
        }
    }

    @Override
    public void close() {}
}

Kafka Streams配置添加拦截器：

Properties streamsProps = new Properties();
// 添加拦截器配置
streamsProps.put(org.apache.kafka.clients.consumer.ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, 
                 "com.yourpackage.OffsetCommitErrorInterceptor");
// 其他Kafka Streams配置...

方案对比 #

• 方案1：实现简单，不用改业务代码，能精准抓取目标日志，但依赖日志格式的稳定性
• 方案2：更贴近底层逻辑，不依赖日志，但要注意Kafka版本兼容性，且异步提交失败的场景可能无法触发异常捕获

优先选方案1，这类错误本身就是通过日志暴露的，拦截日志是最直接可靠的方式。

内容的提问来源于stack exchange，提问作者seke