PubSub Message Handler Async At-least-once Delivery

[nvolynets]

Problem

There is no way to ship the data into GCP PubSub via using com.google.cloud.spring.pubsub.integration.outbound.PubSubMessageHandler & with meeting as follows requirements:

• do it asynchronously (with batching is plus)
• be able programatically achieve finally at-least-once delivery guarantee

To be more descriptive here:

• configure PubSubMessageHandler in async mode

• in this case there is only "publish callback" to be provided for operation status notification

• "publish callback" provides only GCP PubSub message ID as input parameter

• so in "publish callback" original input message is missed + headers

• as result there is no way to correlate original input message (submitted into PubSubMessageHandler) with GCP PubSub response

• as result no way to mark original input (source) message as delivered or redeliver it if something happened

    PubSubMessageHandler adapter = new PubSubMessageHandler(pubsubTemplate, topic);
  
    adapter.setSync(true);
  
    adapter.setPublishCallback(new ListenableFutureCallback<String>() {
        @Override
        public void onFailure(Throwable ex) {
            log.debug("There was an error sending the message", ex);
        }
  
        @Override
        public void onSuccess(String result) {
            log.debug("Message was sent successfully - result: {}", result);
        }
    });
  

And would like to emphasise that trying to achieve at-least-once delivery guarantee. So "sigkill" should be handled by my app without data loss.

Consequently, at the moment custom PubSubMessageHandler implementation is required to achieve what was described above. Or please correct me if I've missed something.

Versions

• spring-cloud.version --> 2020.0.2
• spring-cloud-gcp.version --> 2.0.2

Describe the solution you'd like
Do similar as in org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler:

• add possibility to specify "OutputChannel"
• in a message delegated into "OutputChannel" preserve message headers of original input message

Having this will be a possibility to correlate PubSub response with input source message. And mark/rollback it if needed.

[meltsufin]

@elefeint Can you please take a look when you have a chance?

[elefeint]

@nvolynets Sorry for the delay. The publish callback only returning a single ID string is definitely a problem. At-least-once delivery came up in the past, and our solution was to lead people to synchronous publishing. Blocking code is becoming unacceptable for modern applications, though, as your requirements show.

Out of curiosity, would a more useful callback containing the message headers and/or message itself meet your requirement? I've experimented with it in #482 -- commit 0dc8363 shows what a better callback might look like.

b813efa looks into your proposed solution AbstractReplyProducingMessageHandler as a superclass -- it's the same superclass that HttpRequestExecutingMessageHandler uses, except it uses it synchronously.

The right thing to do here is likely to do both, a more useful callback and the ability to redirect an enhanced response to an output channel.

[nvolynets]

Hi @elefeint ,

Thanks for the feedback. You are correct in that I am looking for.

And yes, synchronous publishing is not an option (verified -> too bad performance).

Going even deeper around an issue with I've bumped (indirectly related with this one described here):

• achieve best performance
• guarantee at-least-once delivery
• preserve ordering

As result looked even into PubSub "batching" feature.

[elefeint]

That makes sense; I'll clean up that pull request and get it in.

To use client library's publish batching, you have to set 4 properties, since any one of them being off will turn batching off:

• spring.cloud.gcp.pubsub.publisher.batching.element-count-threshold
• spring.cloud.gcp.pubsub.publisher.batching.request-byte-threshold
• spring.cloud.gcp.pubsub.publisher.batching.delay-threshold-seconds
• spring.cloud.gcp.pubsub.publisher.batching.enabled