Spring cloud kafka流应用程序因异常而终止

Question

我有一个简单的Spring云kafka流应用程序。每次有异常时，应用程序都会终止，我无法覆盖此行为。当有某些类型的异常时，所需的结果是增量退回，或者继续其他类型的异常。我使用spingCloudVersion-Hoxton.SR3和Spring启动：2.2.6.RELEASE

application.yaml

spring:
 cloud:
   stream:
     binders.process-in-0:
         destination: test

     kafka:
       streams:
         binder:
           deserializationExceptionHandler: logAndContinue
           configuration:
             default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
             default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

豆子

@Bean
public java.util.function.Consumer<KStream<String, String>> process() {
    return input -> input.process(() -> new EventProcessor());
}

@Bean
public StreamsBuilderFactoryBeanCustomizer customizer() {
    return fb -> {
        fb.getStreamsConfiguration().put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
                ContinueOnErrorHandler.class);
    };
}

事件处理器

public class EventProcessor implements Processor<String, String>, ProcessorSupplier<String, String> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(String key, String value) {

        throw new RuntimeException("Some exception");
    }

    @Override
    public void close() {

    }

    @Override
    public Processor<String, String> get() {
        return this;
    }
}

继续错误处理程序

public class ContinueOnErrorHandler implements ProductionExceptionHandler {
    @Override
    public ProductionExceptionHandlerResponse handle(ProducerRecord<byte[], byte[]> record, Exception exception) {
        return ProductionExceptionHandlerResponse.CONTINUE;
}

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

Answer 1

您从消费者处使用的自定义处理器正在进程方法中抛出Runtime Exc的。它不会被任何东西捕获。当抛出该异常时，应用程序只需退出。

您正在使用的生产异常处理程序在这里没有任何效果，因为您在这里没有生产任何内容消费者不生产任何产品。如果您有生产产品的用例，您应该切换到java.util.funciton。改为函数。

为了解决这个问题，当您在自定义处理器(< code>EventProcessor)中处理记录时，如果您得到一个异常，您应该捕捉它并采取适当的操作。例如，这里有一个模板:

        @Override
        public void init(ProcessorContext context) {
            this.context = context;
        }

        @Override
        public void process(String key, String value) {
            try {
                // start processing
                // exception thrown

            }
            catch (Exception e){
                // Take the appropriate action
            }
        }

这样，当处理器中引发异常时，应用程序就不会终止。