subscription.close() leads to "Request payload size exceeds the limit"

[ctavan]

Environment details

• OS: Kubernetes Engine
• Node.js version: 9.3.0
• npm version: –
• @google-cloud/pubsub version: 0.16.2

Steps to reproduce

I have a consumer running on Kubernetes Engine which reads messages from a PubSub subscription and writes them to BigQuery.

I have implemented a clean shutdown logic when my pods get terminated (e.g. during a rolling updated). In that shutdown logic I call subscription.close().

Occasionally calling subscription.close() leads to the following error being emitted in the subscription.on('error') event:

Error: Request payload size exceeds the limit: 524288 bytes.
    at /app/node_modules/grpc/src/client.js:554:15

If I inspect the error that I receive in the subscription.on('error') handler, I can see that the stack trace contains either:

    at Subscription.emit (events.js:159:13)
    at /app/node_modules/@google-cloud/pubsub/src/subscription.js:346:10
    at <anonymous>"    

or:

    at Subscription.emit (events.js:159:13)
    at /app/node_modules/@google-cloud/pubsub/src/subscription.js:869:10
    at <anonymous>

I could trace down the problem as follows:

When subscription.close() is called, this in turn calls flushQueues_():

nodejs-pubsub/src/subscription.js

Lines 381 to 395 in d89287c

	Subscription.prototype.close = function(callback) {
	var self = this;
	this.userClosed_ = true;
	var inventory = this.inventory_;
	inventory.lease.length = inventory.bytes = 0;
	clearTimeout(this.leaseTimeoutHandle_);
	this.leaseTimeoutHandle_ = null;
	this.flushQueues_().then(function() {
	self.closeConnection_(callback);
	});
	};

Within flushQueues_(), in case there are pending acks or nacks, acknowledge_() or modifyAckDeadline_() are being called will all pending ackIds/nackIds respectively:

nodejs-pubsub/src/subscription.js

Lines 631 to 647 in d89287c

	if (acks.length) {
	requests.push(
	this.acknowledge_(acks).then(function() {
	self.inventory_.ack = [];
	})
	);
	}
	if (nacks.length) {
	requests.push(
	this.modifyAckDeadline_(nacks, 0).then(function() {
	self.inventory_.nack = [];
	})
	);
	}
	return Promise.all(requests);

Both these methods are rather similar. In any case they will simply make a call to the API with all the ackIds/nackIds that were passed from flushQueues_().

Here's acknowledge_():

nodejs-pubsub/src/subscription.js

Line 318 in d89287c

connection.write({ackIds}, resolve);

And here's modifyAckDeadline_():

nodejs-pubsub/src/subscription.js

Lines 899 to 905 in d89287c

	connection.write(
	{
	modifyDeadlineAckIds: ackIds,
	modifyDeadlineSeconds: Array(ackIds.length).fill(deadline),
	},
	resolve
	);

Apparently these API-calls are too big, since the corresponding promises reject with an error that gets then emitted and that I then see in my application code:

nodejs-pubsub/src/subscription.js

Lines 323 to 325 in d89287c

	return promise.catch(function(err) {
	self.emit('error', err);
	});

nodejs-pubsub/src/subscription.js

Lines 910 to 912 in d89287c

	return promise.catch(function(err) {
	self.emit('error', err);
	});

The error message mentions 524288 bytes which is precisely 512 KiB.

My suspicion is that the acks/nacks must be sent in batches in case they exceed the maximum payload size for acknowledge/modifyAckDeadline requests?

Unfortunately the REST API docs don't say anything about limits.

In any case I would expect that subscription.close() allows for a clean shutdown, even if there are many pending acks/nacks. The current behavior leads to potential duplicates on my end since pending acks are not correctly being flushed: since the sending of the pending acks itself triggers an error and subsequent crash in my application code the will be re-consumed after the restart of my application, even though they have already been processed.

[callmehiphop]

@ctavan thanks for reporting, I was unaware of any size constraints on these methods, but I would guess your conclusion is correct, in that we need to implement batching.

If you have any spare time and would like to contribute a PR, we'd be very grateful! Otherwise I think I can get to this next week.

[ctavan]

I will try to provide a fix now.

[ctavan]

@callmehiphop I have submitted a potential fix in #65.

I'm simply chunking the ack/modifyAckDeadline requests into chunks of 3000 or less ackIds.

While I was getting the error described in this issue for almost every shutdown of my app with 0.16.2, I haven't seen it with my patch applied so far (and I have provoked ~10 shutdowns with the updated code so far) yet.

Let me know what you think!

[nickcarenza]

the 3000 chunk size might be too much.

error: {
details: "Request payload size exceeds the limit: 524288 bytes."
message: "Failed to "modifyAckDeadline" for 3000 message(s). Reason: 3 INVALID_ARGUMENT: Request payload size exceeds the limit: 524288 bytes."
stack: "Error: Failed to "modifyAckDeadline" for 3000 message(s). Reason: 3 INVALID_ARGUMENT: Request payload size exceeds the limit: 524288 bytes.
    at /usr/src/app/node_modules/@google-cloud/pubsub/build/src/message-queues.js:218:23
    at runNextTicks (node:internal/process/task_queues:59:5)
    at processImmediate (node:internal/timers:437:9)
    at async Promise.all (index 0)
    at async ModAckQueue._sendBatch (/usr/src/app/node_modules/@google-cloud/pubsub/build/src/message-queues.js:221:9)
    at async ModAckQueue.flush (/usr/src/app/node_modules/@google-cloud/pubsub/build/src/message-queues.js:109:13)"
code: 3
}

version 2.10.0