GH-916: MC Binder Producer Error Infrastructure

[garyrussell]

Resolves #916
Resolves #802

• register a pubsub error channel.
• register and subscribe a bridge handler to bridge it to the global error channel.
• pass the error channel to the implementation so it can wire it into the outbound endpoint.
• destroy the infrastructure when unbinding.

Resolves #916

[artembilan]

LGTM

[viniciusccarvalho]

I'm ok with the changes, I hope we can review this on 2.0. I have a few reservations, but for now I'll keep it to myself until we get a change to review the internal guts of AbstractBinder, AbstractMessageChannelBinder and Consumer/ProducerDestination classes

[garyrussell]

Hold off merging; I will add some tests.

[sobychacko]

@garyrussell how do these changes correlate to when someone set something like spring.cloud.stream.bindings.error.destination=error-dest. In that case, I think we already create a bridge handler and subscribe to the error channel and send any errors from the error channel to the destination. Referring to this recent commit: 22141a8.

It may not be related, but I am just curious.

[garyrussell]

@sobychacko Interesting; I wasn't aware of that. They're (sort of) related, but not really.

Currently, the binders don't send errors to the Spring Integration global errorChannel.

It looks to me like that code binds the global errorChannel to a destination (enabling errors to be sent to a binder destination). This would only kick in if the user app publishes to the error channel somehow.

This PR registers an error channel for (async) errors for producer destinations. (And this one, which is already merged, does the same for consumer destinations). On the consumer side, this publication is performed regardless. DLQ processing now consumes from that channel before sending to a DLQ. On the producer side, the error channel is disabled by default because this adds overhead (in the case of Rabbit, but not Kafka).

In both cases, those error channels are bridged to the global errorChannel.

This allows the user to consume producer and consumer errors, either individually or collectively (with a normal SI flow).

Since the global errorChannel is a pubsub channel, these errors will also go to that binding as well (spring.cloud.stream.bindings.error.destination=error-dest), if present.

[sobychacko]

Thanks, @garyrussell for the explanation. That code actually binds a bridged channel from the error channel to the destination. The bridge channel goes through our normal channel configuration for content type conversion etc.

[viniciusccarvalho]

Merged via 0eab8a5