Cfrgtkky

How to do the error handling for the reactive stream pipeline. Like

• Application Error Handling (ex: errorChannel)


• System Error Handling (working with DLQ, reprocessing etc)

The current documentation only describes error handling for the non-reactive pipeline.
https://docs.spring.io/spring-cloud-stream/docs/Fishtown.BUILD-SNAPSHOT/reference/htmlsingle/#_application_error_handling

Spring cloud stream has done pretty easy for the users by providing simple configs for the error handling scenarios. It would be great if the same error handling usecases (with same configs) work for the reactive stream pipeline as well. The use cases and the respective config details as below:

• @StreamListner("errorChannel") annotation for the global error handling


• @KafkaListener(id="bar", topics = "reactive-stream-error-topic")


• Configs for the DLQ and producing failed messages to the error topics
  spring.cloud.stream.kafka.bindings.input.consumer.enableDlq=true
  spring.cloud.stream.kafka.bindings.input.consumer.dlqName=reactive-stream-error-topic
  

The example from the documentation works fine with spring-cloud-stream but the same thing gives an error for reactive pipeline. Any guidelines in this direction would be of great help for the community. Thanks in advance!

@SpringBootApplication

@EnableBinding(Sink.class)

public class ReactiveStreamSinkApplication {



public static void main(String args) {

    SpringApplication.run(ReactiveStreamSinkApplication.class, args);

}



@StreamListener

public void receive(@Input(Sink.INPUT) Flux<String> inputFlux) {

    inputFlux.subscribe(System.out::println);

    throw new RuntimeException("BOOM!");

}



@StreamListener("errorChannel")

public void error(Message<?> message) {

    // log the error msg

    System.out.println("Handling ERROR: " + message);

}



@KafkaListener(id="bar", topics = "reactive-stream-error-topic")

public void error(String in) {

    System.out.println(in + " from DLQ");

}

}

Sorry for the late reply.

First, you have a problem in your code and a way you are treating how exception may be thrown from the reactive stream. Basically you are dealing with a declarative handler which is treated quite different. In your code the receive method will be invoked only once during the startup and initialization. So throwing exception from it is not at all equal to the exception thrown during the stream processing, which is the error handling mechanism you are inquiring about was designed for. But. . .

That aside. . .

With introduction of Spring Cloud Function programming model we are considering shifting attention from reactive module all together since Spring Cloud Function already provides support for reactive programming model.
So consider the following:

@SpringBootApplication

@EnableBinding(Sink.class)

public class ReactiveStreamSinkApplication {



    public static void main(String args) {

        SpringApplication.run(ReactiveStreamSinkApplication.class,

        "--spring.cloud.stream.function.definition=myconsumer");

    }



    @Bean

    public Consumer<Flux<String>> myconsumer() {

        return stream -> stream.subscribe(value -> {

            if ("foo".equals(value)) {

                throw new RuntimeException("BOOM!");

            }

            System.out.println("Received value: " + value);

        });

    }



    @StreamListener("errorChannel")

    public void error(Message<?> message) {

        // log the error msg

        System.out.println("Handling ERROR: " + message);

    }

}

Try that and let us know.