Messaging - swallowed exception from CS

[danielkec]

Thanks @santhoshpt for finding the bug, signature INCOMING_COMPLETION_STAGE_2_MSG swallows CS exception.

	@Incoming("fromKafka")
	public void getMessages(Message<String> message) {
		return CompletableFuture.failedFuture(new RuntimeException());
	}

[danielkec]

Workaround:

    @Incoming("fromKafka")
    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
    public SubscriberBuilder<Message<String>, Void> getMessages() {
        return ReactiveStreams.<Message<String>>builder()
                .map(msg -> {
                     return CompletableFuture.failedFuture(new RuntimeException());
                })
                .flatMap(ReactiveStreams::fromCompletionStageNullable)
                .onError(t -> LOG.log(Level.SEVERE, "Stream killed by error", t))
                .ignore();
    }