Receiving lots of duplicate messages

[jamessharp]

Hi

I've been getting a lot of duplicate messages recently, specifically when running with a largeish backlog of messages, limited maxMessages and delayed acking. I have downgraded to v0.13 (i.e. before StreamingPull) and the dupes have disappeared.

This could possibly be related to #68, googleapis/google-cloud-java#2465 and googleapis/google-cloud-go#778

Environment details

• OS: ubuntu linux (running under GKE)
• Node.js version: 9.4.0
• @google-cloud/pubsub version: 0.16.5
• subscription ack deadline: 600s

Steps to reproduce

The following reproduces the issue fairly reliably in a container on GKE, though the number of duplicates we were seeing in our actual usage was higher than this seems to produce.

publisher

const NOTIFICATION_TOPIC = 'test';
const pubsub = require('@google-cloud/pubsub')()
const PQueue = require('p-queue');

const publisher = pubsub.topic(NOTIFICATION_TOPIC).publisher();

// The queue ensures there are never more than 500 publishes in flight at
// any one time
const queue = new PQueue({ concurrency: 500 });

function publish() {
  return publisher.publish(Buffer.from('test'));
}

const proms = [];
for (let ii = 0; ii < 2000; ii++) {
  proms.push(queue.add(publish));
}

Promise.all(proms).then(() => console.log('done')).catch(err => console.error(err));

subscriber

const NOTIFICATION_TOPIC = 'test';
const NOTIFICATION_SUBSCRIPTION = 'test-test';

const pubsub = require('@google-cloud/pubsub')();
const ackHistory = new Map();

const topic = pubsub.topic(NOTIFICATION_TOPIC);
const subscription = topic.subscription(NOTIFICATION_SUBSCRIPTION, {
  flowControl: {
    maxMessages: 25
  }
});

var count = 0;
subscription.on('message', message => {
    var ackCount = ackHistory.get(message.ackId) || 0;
    ackCount++;
    ackHistory.set(message.ackId, ackCount);
    if (ackCount > 1) {
      console.log('Already seen this ackId',  message.ackId, ackCount);
    }

    // In our code we went and called some internal services. Simulating this
    // by using a timeout
    setTimeout(() => {
      message.ack();
      console.log('ACKED', count++);
    }, 300);
});

Let me know if I can provide any more information

[stephenplusplus]

Thanks for reporting. Do you think this is a duplicate of #68?

[jamessharp]

It could be, but I haven't noticed any regularity (e.g. the 15 min spikes) like in #68. The dupes come pretty quickly (the repro scenario above only takes a couple of mins to run anyway). It feels like in some cases the modifyAckDeadline isn't getting back to the pubsub in time (the initial lease is for 10s I think)

[callmehiphop]

@jamessharp we've made some adjustments to the flow control which I think should help cut down on duplicate messages. If you have the bandwidth to test #92 that would be fantastic!

[jamessharp]

It feels like it's happening less frequently, but I've just run the above code (putting 10000 messages in rather than 2000) and ended up handling 10875 (i.e. 875 were duplicated).

The subscription had an ack deadline of 600s and the first dupes came in after around 40s

Also, for the record, getting the following console log appearing every 15s or so:

Error: 14 UNAVAILABLE: The service was unable to fulfill your request. Please try again. [code=8a75]
    at createStatusError (/service-notifications/node_modules/grpc/src/client.js:64:15)
    at ClientDuplexStream._emitStatusIfDone (/service-notifications/node_modules/grpc/src/client.js:270:19)
    at ClientDuplexStream._receiveStatus (/service-notifications/node_modules/grpc/src/client.js:248:8)

[callmehiphop]

@jamessharp I merged a pr (#99) that should help cut down on the number of duplicates. However it doesn't appear to completely fix the issue. I'm going to continue to monitor this and any new info can be tracked via #83. Thanks again for opening this issue and testing our changes!

[yonran]

As I commented on googleapis/google-cloud-java#2465, The StreamingPull server-side API has a bug that makes it unusable for topics containing small messages (such as GCS Object Change Notifications). If the caller of StreamingPull does not process messages quickly enough, then these messages will be duplicated and the subscriber will keep falling behind. Even if you Ack every message that arrives to the client, PubSub duplicates these messages because the client didn’t acknowledge other messages in time that the server didn’t even give to the client.

I think a rule of thumb that might work is, if there ever is a backlog and a single subscriber cannot clear the entire backlog within an hour, then don’t use StreamingPull; use Pull instead.