如何在Kafka Stream中忽略主题分区内存在Schema错误的特定偏移量？

当然可以！完全能在复制生产主题消息到测试主题的过程中跳过那些带有无效Schema ID的偏移量。针对你遇到的这个反序列化报错问题，这里有几个实用的解决方案：

方案1：用Kafka Streams配置异常处理器 #

如果你们是通过Kafka Streams来做主题复制，最简单的方式是配置内置的异常处理器，让它自动跳过坏消息：

在你的Streams配置文件中添加以下参数：

default.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndContinueExceptionHandler

这个处理器会把错误日志记录下来，然后直接跳过那个有问题的偏移量，继续处理下一条消息。

要是你需要更精准的控制（比如只跳过“Schema not found”这类特定错误），可以自定义一个异常处理器类：

public class SkipSchemaNotFoundHandler implements DeserializationExceptionHandler {
    @Override
    public DeserializationHandlerResponse handle(ProcessorContext context, ConsumerRecord<byte[], byte[]> record, Exception exception) {
        // 只跳过Schema Registry返回404的情况
        if (exception.getCause() instanceof RestClientException && ((RestClientException) exception.getCause()).getStatus() == 404) {
            context.logger().error("Skipping invalid record at offset {} (topic: {}, partition: {}) - Schema not found",
                    record.offset(), record.topic(), record.partition(), exception);
            return DeserializationHandlerResponse.CONTINUE;
        }
        // 其他异常还是抛出，避免错过其他问题
        return DeserializationHandlerResponse.FAIL;
    }

    @Override
    public void configure(Map<String, ?> configs) {}
}

然后在配置里指定这个自定义类：

default.deserialization.exception.handler=com.yourcompany.SkipSchemaNotFoundHandler

方案2：用Kafka MirrorMaker 2复制主题 #

如果你们用MirrorMaker 2来做跨主题复制，可以通过配置源消费者的参数来跳过坏消息：

在MirrorMaker 2的MirrorSourceConnector配置中添加：

source.consumer.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
source.consumer.key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
source.consumer.value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

这里用ByteArrayDeserializer先把消息以原始字节的形式读取，绕过Avro Schema验证，先把所有消息复制到测试主题。之后你可以在测试主题上单独处理，过滤掉那些无法解析的坏消息。

方案3：在Kafka Connect中配置错误容忍 #

如果你们是用Kafka Connect直接将消息写入HDFS，同时需要复制到测试主题，可以在源连接器中开启错误容忍配置：

给你的Kafka Connect源连接器添加以下参数：

errors.tolerance=all
errors.log.enable=true
errors.log.include.messages=true

errors.tolerance=all会让连接器自动跳过处理失败的记录，继续处理后续的正常消息，这样就能同时完成主题复制和HDFS写入，并且跳过那些带有无效Schema ID的消息。

临时应急方案：手动重置偏移量 #

如果你们已经明确知道坏消息的具体偏移量（比如你日志里的partition=9, offset=1259032），可以用Kafka的命令行工具手动重置消费者组的偏移量，直接跳过这个位置：

kafka-consumer-groups.sh --bootstrap-server your-kafka-broker:9092 --group your-consumer-group-id --topic xxxx:9 --reset-offsets --to-offset 1259033 --execute

这个方法适合少量已知的坏消息，要是坏消息数量多，还是用上面的自动处理方案更高效。

你遇到的错误日志：

Exception in thread "StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Failed to deserialize value for record. topic=xxxx, partition=9, offset=1259032
Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id 600
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema not found
io.confluent.rest.exceptions.RestNotFoundException: Schema not found
io.confluent.rest.exceptions.RestNotFoundException: Schema not found

本质是消息携带的Schema ID 600在当前使用的Schema Registry中不存在，导致反序列化失败。上面的方案都能帮你绕过这些坏消息，正常处理其他有效记录。

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