Subscriber with flow control blocked by unacked messages

[ncjones]

Environment details

• OS: OSX / GKE + Alpine Linux v3.6
• Node.js version: 8.9.3
• npm version: 5.5.1
• @google-cloud/pubsub version: 0.16.2

Steps to reproduce

1. Create subscriber with flow control max messages of 1 and a handler that throws an exception.
2. Send two messages - first message results in error as expected but second message is never handled and first message never retried.

In our scenario the flow control is currently set to max messages of 8 and there are two running instances of the application. As problematic messages arrive we see the message ack rate drop off in Stackdriver until it reaches 0/s (I assume after the 16th unhandled error). When the application is restarted it will resume processing messages and repeat the same pattern and eventually get stuck again. We also see a huge number of pull operations (~4k/s) even though the subscriber is not logging any activity.

I can workaround the issue by explicitly nacking messages which have uncaught errors but (correct me if I'm wrong) this means the problematic message gets redelivered immediately instead of after the acknowledgement deadline. The delayed redelivery is nice since it reduces the noise in the logs when encountering a bad message or temporary network issue etc.

I have tried testing locally with the emulator and version 0.18.0 of the client library I still get the same issues.

[callmehiphop]

@ncjones this sounds reasonable to me. What if we allowed an (optional) re-delivery time to be passed into nack.

message.nack(60); // re-deliver in 60 seconds

@stephenplusplus @alexander-fenster any thoughts here?

[ncjones]

@callmehiphop nack with delay would be useful but really I want to make sure that unacked messages get redelivered and don't block the subscriber (it's hard to guarantee message handlers will always explicitly ack or nack a message).

Any idea why unacked messages are blocking my subscriber? Here is how I create my subscribers (handler.handleMessage returns a promise):

    function startSubscriber({ subscriptionName, handler}) {
      logger.info(`subscribing to ${subscriptionName}`)
      const subscription = pubsubClient.subscription(subscriptionName, {
        flowControl: {
          maxMessages: 8,
        }
      })
      subscription.on('message', msg => {
        return handler.handleMessage(msg)
          .catch(e => {
            logger.error(e)
            // catch-all msg.nack() would go here but is unreliable since
            // msg handler may have already handled error without ack/nack
          })
      })
    }

[callmehiphop]

We automatically extend the ack deadline of unacked messages, so if you don't explicitly ack or nack messages they will (in theory) just linger around forever. This can block new messages from being received because the subscriber thinks you are still processing the unacked messages. As the code is today, your best bet is to nack any messages you want redelivered.

[callmehiphop]

We have a release coming up in the next several days that will introduce a new option called maxExtension, this will release locks on any messages that have been in memory past the specified limit. You should be able to use this option to specify when you want messages that you may have already handled redelivered.

const subscription = topic.subscription('my-sub', {
  flowControl: {
    maxExtension: 60 * 5 // 5 minutes
  }
});