Excessive message re-delivery when trying to consume batches of pubsub data

[Navina Ramesh]

Hi ,

Hi, I am noticing a lot of message re-deliveries in my pubsub consumer application, even though I have configured max ack deadline for subscription (600 seconds). I am using the async client api for pull-based subscriptions. 

Here is an example of the redelivery from my logs:

```

xxx@XXX-MacBook-Pro Downloads % grep 4352639842273916 ~/Downloads/a.txt 
2022/04/08 15:07:01.123 DEBUG [PubsubConsumerImpl] [Gax-11] #### RECEIVE : 4352639842273916
2022/04/08 15:07:01.123 DEBUG [PubsubConsumerImpl] [Gax-11] #### QUEUED : 4352639842273916
#### RETURNED : 33 : 4352639842273916
2022/04/08 15:07:05.332 DEBUG [PubsubMessageQueue] [github_events__0__0__20220408T2206Z] ### ACK 33 : 4352639842273916
2022/04/08 15:07:05.333 DEBUG [PubsubMessageQueue] [github_events__0__0__20220408T2206Z] ### clear ack 33:4352639842273916
#### 33 : 4352639842273916
2022/04/08 15:07:07.043 DEBUG [PubsubConsumerImpl] [Gax-13] #### RECEIVE : 4352639842273916
2022/04/08 15:07:07.043 WARN [PubsubConsumerImpl] [Gax-13] #### DUPLICATE : 4352639842273916
2022/04/08 15:07:57.352 DEBUG [PubsubConsumerImpl] [Gax-29] #### RECEIVE : 4352639842273916
2022/04/08 15:07:57.352 DEBUG [PubsubConsumerImpl] [Gax-29] #### QUEUED : 4352639842273916
#### RETURNED : 81 : 4352639842273916
2022/04/08 15:07:57.406 DEBUG [PubsubMessageQueue] [github_events__0__1__20220408T2207Z] ### ACK 81 : 4352639842273916
2022/04/08 15:07:57.407 DEBUG [PubsubMessageQueue] [github_events__0__1__20220408T2207Z] ### clear ack 81:4352639842273916
#### 81 : 4352639842273916
2022/04/08 15:08:49.184 DEBUG [PubsubConsumerImpl] [Gax-41] #### RECEIVE : 4352639842273916
2022/04/08 15:08:49.184 DEBUG [PubsubConsumerImpl] [Gax-41] #### QUEUED : 4352639842273916
#### RETURNED : 130 : 4352639842273916
2022/04/08 15:08:49.237 DEBUG [PubsubMessageQueue] [github_events__0__2__20220408T2208Z] ### ACK 130 : 4352639842273916
2022/04/08 15:08:49.237 DEBUG [PubsubMessageQueue] [github_events__0__2__20220408T2208Z] ### clear ack 130:4352639842273916
#### 130 : 4352639842273916
```

If you look at the line with “DUPLICATE”, it is getting re-delivered within 6 seconds and also, through a different grpc thread (Gax-11 and Gax-13) . Why is the same message being delivered to 2 different threads?

I am using a subscription with 600 seconds for ackDeadline. So, the messages shouldn’t get re-delivered unless the ack didn’t reach the server ? is that something that can happen?

Fyi, in my application, I am trying to consume a batch of messages and ack them all when the processing is complete. I verified that my code is invoking ACK on all the processed messages. I am using flow control to manage the batch of consumed messages. But it doesn't seem to improve the duplicates situation.

Any help would be appreciated! 

Thanks!
Navina

TLDR; Usage pattern of my application:
```Consumer implementation:

start():

• Create a subscription

• seek to a known snapshot (if any); if not found, create a new snapshot

• Start one subscriber (message receiver adds the received messages to a shared queue) 

 

fetch(): 

• ACK previous batch of messages (if any) 

• Poll a batch of messages from shared queue and return 

 

checkpoint(): 

• ACK any processed messages

• Create a new snapshot 

 

close():

• ACK any previous batch of messages that are pending ack 

• Take a snapshot

• stop subscriber 

• delete subscription

 

Application:
start()

While (some critteria is reached): {
        fetch() 
}

checkpoint()

close()
```

[Navina Ramesh]

Also, is it possible that my subscriber is not shutting down? 

I use `subscriber.stopAsync().awaitTerminated(30, TimeUnit.SECONDS)` and majority of the time, I see a stacktrace saying the subscriber was not closed. It's state is `STOPPING`. COuld this lead to duplicates??
Increasing the timeout to 120 seconds also doesn't seem to help. 

[dikaur]

Hello,

Re-delivery of the messages in some cases  can happen when client tasks are unhealthy as code might not see the message before the ack deadline expires. 

The possible solution can be to reduce the queue size of the number of messages pulled by subscribers.

You can also refer to the solution provided in [1] [2]  for the flow control of the messages.

[1] https://stackoverflow.com/questions/44812934/google-pubsub-flow-control

[2] https://cloud.google.com/pubsub/docs/pull#flow_control