Ack pubSub message outside of the MessageReciever

Question

I am using async Pull to pull messages from a pupSub topic, do some processing and send messages to ActiveMQ topic.

With the current configuration of pupSub I have to ack() the messages upon recieval. This however, does not suit my use case, as I need to ONLY ack() messages after they are successfully processed and sent to the other Topic. this means (per my understanding) ack()ing the messages outside the messageReciver.

I tried to save the each message and its AckReplyConsumer to be able to call it later and ack() the messages, this however does not work as expected. and not all messages are correctly ack() ed.

So I want to know if this is possible at all. and if Yes how

my subscriber configs

 public Subscriber getSubscriber(CompositeConfigurationElement compositeConfigurationElement, Queue<CustomPupSubMessage> messages) throws IOException {

    ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(compositeConfigurationElement.getPubsub().getProjectid(),
            compositeConfigurationElement.getSubscriber().getSubscriptionId());

    ExecutorProvider executorProvider =
            InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(2).build();

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
            (PubsubMessage message, AckReplyConsumer consumer) -> {
                messages.add(CustomPupSubMessage.builder().message(message).consumer(consumer).build());
            };

    // The subscriber will pause the message stream and stop receiving more messages from the
    // server if any one of the conditions is met.
    FlowControlSettings flowControlSettings =
            FlowControlSettings.newBuilder()
                    // 1,000 outstanding messages. Must be >0. It controls the maximum number of messages
                    // the subscriber receives before pausing the message stream.
                    .setMaxOutstandingElementCount(compositeConfigurationElement.getSubscriber().getOutstandingElementCount())
                    // 100 MiB. Must be >0. It controls the maximum size of messages the subscriber
                    // receives before pausing the message stream.
                     .setMaxOutstandingRequestBytes(100L * 1024L * 1024L)
                    .build();


    //read credentials
    InputStream input = new FileInputStream(compositeConfigurationElement.getPubsub().getSecret());
    CredentialsProvider credentialsProvider = FixedCredentialsProvider.create(ServiceAccountCredentials.fromStream(input));


    Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver)
                                    .setParallelPullCount(compositeConfigurationElement.getSubscriber().getSubscriptionParallelThreads())
                                    .setFlowControlSettings(flowControlSettings)
                                    .setCredentialsProvider(credentialsProvider)
                                    .setExecutorProvider(executorProvider)
                                    .build();

   return subscriber;
}

my processing part

            jmsConnection.start();
            for (int i = 0; i < patchSize; i++) {
                var message = messages.poll();
                if (message != null) {
                    byte[] payload = message.getMessage().getData().toByteArray();
                    jmsMessage = jmsSession.createBytesMessage();
                    jmsMessage.writeBytes(payload);
                    jmsMessage.setJMSMessageID(message.getMessage().getMessageId());

                    producer.send(jmsMessage);
                    list.add(message.getConsumer());
                } else break;
            }

            jmsSession.commit();
            jmsSession.close();
            jmsConnection.close();
            // if upload is successful then ack the messages
            log.info("sent " + list.size() + " in direction " + dest);
            list.forEach(consumer -> consumer.ack());

Answer 1

There is nothing that requires messages to be acked within the MessageReceiver callback and you should be able to acknowledge messages asynchronously. There are a few things to keep in mind and look for:

1. Check to ensure that you are calling ack before the ack deadline expires. By default, the Java client library does extend the ack deadline for up to 1 hour, so if you are taking less time than that to process, you should be okay.
2. If your subscriber is often flow controlled, consider reducing the value you pass into setParallelPullCount to 1. The flow control settings you pass in are passed to each stream, not divided among them, so if each stream is able to receive the full value passed in and your processing is slow enough, you could be exceeding the 1-hour deadline in the client library without having even received the message yet, causing the duplicate delivery. You really only need to use setParallelPullCount to a larger value if you are able to process messages much faster than a single stream can deliver them.
3. Ensure that your client library version is at least 1.109.0. There were some improvements made to the way flow control was done in that version.
4. Note that Pub/Sub has at-least-once delivery semantics, meaning messages can be redelivered, even if ack is called properly. Note that not acknowledging or nacking a single message could result in the redelivery of all messages that were published together in a single batch. See the "Message Redelivery & Duplication Rate " section of "Fine-tuning Pub/Sub performance with batch and flow control settings."

If all of that still doesn't fix the issue, then it would be best to try to create a small, self-contained example that reproduces the issue and open up a bug in the GitHub repo.