Periodically getting unavailable response codes for streaming pulls and expired for acks #411

Environment details

1. Pubsub
2. OS type and version: ubuntu:20.04
3. Java version: openjdk:11.0.9
4. pubsub version(s): google-cloud-pubsub lib, 1.103.1 or 1.108.6 version

Steps to reproduce

1. Create subscription with ackDeadline=600 seconds
2. Publish solid number of messages (in my case it was ~20k stale messages with 1 rps increase)
3. Create a com.google.cloud.pubsub.v1.Subscriber with 8 outstanding elements that is instantly putting incoming messages into queue and queue consumer that is consuming and acknowledging messages from the queue with 2 rps throughput:

val queue = ...
Subscriber
    .newBuilder(
        subscription,
        (message, ackCallback) => queue.offer(message -> ackCallback)
    )
    .setFlowControlSettings(FlowControlSettings.newBuilder()
        .setMaxOutstandingElementCount(8)
        .setLimitExceededBehavior(LimitExceededBehavior.Block)
        .build()
    )
    .build()
    .startAsync()

4. Monitor stackdriver charts during one hour:

• acknowledge message operations chart should have ~45 min periods, where during ~15 first mins all acks are success and ~30 following mins all acks are expired:

• streaming pull message operations chart should have unavailable response codes each ~45 mins:

Hope for your attention. Thanks!