如何跳过 KafkaStreams API 中的 Avro 序列化异常？

Question

我有一个由 KafkaStreams Java api 编写的 Kafka 应用程序。它从 Mysql binlog 中读取数据并做一些与我的问题无关的事情。问题是一个特定的行在从 avro 反序列化时产生错误。我可以深入研究 Avro 模式文件并找到问题，但总的来说，我需要的是一个宽容的异常处理程序，在遇到此类错误时不会使整个应用程序停止。这是我的流应用程序的主要部分：

StreamsBuilder streamsBuilder = watchForCourierUpdate(builder);

        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);
        kafkaStreams.start();
        Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
    }

    private static StreamsBuilder watchForCourierUpdate(StreamsBuilder builder){
        CourierUpdateListener courierUpdateListener = new CourierUpdateListener(builder);
        courierUpdateListener.start();
        return builder;
    }

    private static Properties configProperties(){

        Properties streamProperties = new Properties();

        streamProperties.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, Configs.getConfig("schemaRegistryUrl"));
        streamProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, "courier_app");
        streamProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Configs.getConfig("bootstrapServerUrl"));
        streamProperties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
        streamProperties.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/state_dir");
        streamProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "3");
        streamProperties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
        streamProperties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
        streamProperties.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG");
        streamProperties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
                CourierSerializationException.class);

        return streamProperties;

    }

这是我的 CourierSerializationException class:

public class CourierSerializationException implements ProductionExceptionHandler {
    @Override
    public ProductionExceptionHandlerResponse handle(ProducerRecord<byte[], byte[]> producerRecord, Exception e) {
        Logger.logError("Failed to de/serialize entity from " + producerRecord.topic() + " topic.\n" + e);
        return ProductionExceptionHandlerResponse.CONTINUE;
    }

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

    }
}

不过，只要发生 avro 反序列化异常，流就会关闭，应用程序也不会继续。我是不是漏掉了什么！

Answer 1

你试过用kafka提供的default.deserialization.exception.handler吗？您可以使用 LogAndContinueExceptionHandler 来记录并继续。

我可能错了，但我认为通过实现 ProductionExceptionHandler 创建 Customexception 仅适用于 kafka 端的网络相关错误。

将其添加到属性中，看看会发生什么：

> props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);

如何跳过 KafkaStreams API 中的 Avro 序列化异常？

How to skip an Avro serialization exception in KafkaStreams API?

java

avro

apache-kafka

apache-kafka-streams