googleapis / googleapis/google-cloud-node
Messages not emitted in time for processing, resulting in failed ack deadline extensions and redeliveries
- Dominant language
- TypeScript
- Stars
- 3.2k
- Forks
- 712
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 99
Description
### Description
Hello, I've been experiencing message redeliveries when using a subscription with exactly once delivery enabled where batches of messages are published. It is not clear whether this is a client or product issue, therefore I first wanted to make sure there is no misunderstanding/misconfiguration on my side.
#### Environment details
- OS: MacOS 13
- Node.js version: 20.8.1
- npm version: 8.1.0
- `@google-cloud/pubsub` version: v4.0.5
#### Steps to reproduce
1. Create a pull subscription with exactly once delivery enabled and ack deadline of 60 seconds
e.g.
```
const projectId = "some-project-id";
const topicNameOrId = "test-topic";
const pubsub = new PubSub({ projectId });
const topic = pubsub.topic(topicNameOrId);
const subscriptionName = "eod-with-batching";
await topic.createSubscription(subscriptionName, {
enableExactlyOnceDelivery: true,
ackDeadlineSeconds: 60,
});
```
2. Set up a subscription with the following configuration (I would like to manually extend the lease)
```
const maxMessages = 300; // a couple of hundred should suffice
const subscription = topic.subscription(subscriptionName, {
flowControl: {
allowExcessMessages: false,
maxMessages: maxMessages,
maxExtensionMinutes: 0,
},
maxAckDeadline: Duration.from({
seconds: 90,
}),
minAckDeadline: Duration.from({
seconds: 90,
}),
streamingOptions: {
maxStreams: 2, // issue more prominent with more than one stream
},
});
```
3. Add a message handler which adds an interval which extends the deadline of each message received every 60 seconds by 120 seconds and awaits the completion of a long-running async action.
```
const waitFor = (ms: number) =>
new Promise((resolve) => {
setTimeout(() => {
resolve(0);
}, ms);
});
const messageMap: Record = {};
subscription.on("message", async (message: Message) => {
if (messageMap[message.id]) {
console.log(`message ${message.id} unexpectedly redelivered. delivery attempt:${message.deliveryAttempt}`);
}
messageMap[message.id] = message;
const interval = setInterval(async () => {
try {
await message.modAckWithResponse(120);
} catch (e) {
console.log("error", e);
}
}, 1000 * 60);
await waitFor(1000 * 60 * 1200);
});
```
4. Create and publish messages (note: redeliveries occur with or without batching)
```
const messagesToPublish: string[] = [];
// ensure some messages are in the backlog
const numberOfBatchedMessages = maxMessages * 3;
for (let i = 0; i < numberOfBatchedMessages; i++) {
messagesToPublish.push(
`Test message ${i + 1} at ${new Date().toISOString()}`
);
}
const publishOptions: PublishOptions = {
batching: {
maxMessages: 5,
maxMilliseconds: 100,
},
};
const batchPublisher = pubsub.topic(topicNameOrId, publishOptions);
const promises = messagesToPublish.map((m) => {
return batchPublisher.publishMessage({
data: Buffer.from(m),
});
});
const publishedMessages = await Promise.all(promises);
console.log(`published ${publishedMessages.length} messages`);
```
5. Wait for redeliveries starting to happen within minutes of running the example above. Note that the issue is intermittent.
As I understand the documentation here for EOD (https://cloud.google.com/pubsub/docs/exactly-once-delivery), as long as
1) message is being processed
2) message deadline keeps being extended
3) message is not negatively acked
the message should not get redelivered (but it is).
Is there something I am missing/doing incorrectly or this is expected behaviour?
Contributor guide
Assessment
This issue has not been assessed yet.