googleapis / googleapis/google-cloud-node

Messages not emitted in time for processing, resulting in failed ack deadline extensions and redeliveries

Open
#7,632 8 comments 0 reactions 1 assignee Claimed by @feywind View on GitHub
api: pubsub priority: p3 status: investigating type: question
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.