I'm trying to configure a Pub/Sub subscription with a dead-letter topic (cf. https://cloud.google.com/pubsub/docs/dead-letter-topics) in order to limit the number of times the message gets redelivered when it gets nack'd. To this end, I've created the following example program:
package main
import (
"context"
"flag"
"fmt"
"log"
"os"
"time"
"cloud.google.com/go/pubsub"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
const (
topicID = "my-topic"
deadLetterTopicID = "my-dead-letter-topic"
subscriptionID = "my-subscription"
)
var (
pubsubEmulatorHost string
projectID string
)
func main() {
flag.StringVar(&pubsubEmulatorHost, "pubsubEmulatorHost", "", "Pub/Sub emulator host (e.g. localhost:8085)")
flag.StringVar(&projectID, "projectID", "my-project", "Google Project ID")
flag.Parse()
if pubsubEmulatorHost != "" {
os.Setenv("PUBSUB_EMULATOR_HOST", pubsubEmulatorHost)
defer os.Unsetenv("PUBSUB_EMULATOR_HOST")
}
client, err := pubsub.NewClient(context.Background(), projectID)
if err != nil {
log.Fatalf("NewClient: %v", err)
}
topic, err := client.CreateTopic(context.Background(), topicID)
if err != nil {
if status.Code(err) == codes.AlreadyExists {
topic = client.Topic(topicID)
log.Printf("Topic %s already exists", topicID)
} else {
log.Fatalf("CreateTopic: %v", err)
}
}
defer func() {
topic.Stop()
if err := topic.Delete(context.Background()); err != nil {
log.Fatalf("Delete topic: %v", err)
}
}()
deadLetterTopic, err := client.CreateTopic(context.Background(), deadLetterTopicID)
if err != nil {
if status.Code(err) == codes.AlreadyExists {
deadLetterTopic = client.Topic(deadLetterTopicID)
log.Printf("Topic %s already exists", deadLetterTopicID)
} else {
log.Fatalf("CreateTopic: %v", err)
}
}
defer func() {
deadLetterTopic.Stop()
if err := deadLetterTopic.Delete(context.Background()); err != nil {
log.Fatalf("Delete dead-letter topic: %v", err)
}
}()
sub, err := client.CreateSubscription(context.Background(), subscriptionID, pubsub.SubscriptionConfig{
Topic: topic,
DeadLetterPolicy: &pubsub.DeadLetterPolicy{
DeadLetterTopic: fmt.Sprintf("projects/%s/topics/%s", projectID, deadLetterTopicID),
MaxDeliveryAttempts: 5,
},
})
if err != nil {
log.Fatalf("CreateSubscription: %v", err)
}
defer func() {
if err := sub.Delete(context.Background()); err != nil {
log.Fatalf("Delete subscription: %v", err)
}
}()
go func() {
sub.Receive(context.Background(), func(ctx context.Context, msg *pubsub.Message) {
log.Printf("Got message %q upon delivery attempt %d", msg.Data, msg.DeliveryAttempt)
msg.Nack()
})
}()
result := topic.Publish(context.Background(), &pubsub.Message{Data: []byte("Hello, world!")})
messageID, err := result.Get(context.Background())
if err != nil {
log.Fatalf("Get message ID of publish call: %v", err)
}
log.Printf("Published message with ID %s", messageID)
time.Sleep(20 * time.Second)
}
The script runs in two modes, one with a genuine Pub/Sub project (called my-project
here) and one using the GCloud Pub/Sub emulator by setting the PUBSUB_EMULATOR_HOST
environment variable. I would expect, given that the subscription's DeadLetterPolicy
has MaxDeliveryAttempts
set to 5, that the nack'd Pub/Sub message is delivered approximately 5 times (the docs indicate that this is a best effort). If I run the script on a real Pub/Sub project, however, I get the following output:
> go run main.go
2020/06/22 23:59:37 Published message with ID 1294186248588871
2020/06/22 23:59:38 Got message "Hello, world!" upon delivery attempt 824637866440
2020/06/22 23:59:40 Got message "Hello, world!" upon delivery attempt 824634417896
2020/06/22 23:59:41 Got message "Hello, world!" upon delivery attempt 824634418592
2020/06/22 23:59:43 Got message "Hello, world!" upon delivery attempt 824637866928
2020/06/22 23:59:44 Got message "Hello, world!" upon delivery attempt 824638981864
2020/06/22 23:59:45 Got message "Hello, world!" upon delivery attempt 824640667960
2020/06/22 23:59:47 Got message "Hello, world!" upon delivery attempt 824634418712
2020/06/22 23:59:49 Got message "Hello, world!" upon delivery attempt 824638982160
2020/06/22 23:59:50 Got message "Hello, world!" upon delivery attempt 824640667760
2020/06/22 23:59:51 Got message "Hello, world!" upon delivery attempt 824634418000
2020/06/22 23:59:52 Got message "Hello, world!" upon delivery attempt 824633942168
2020/06/22 23:59:53 Got message "Hello, world!" upon delivery attempt 824633942712
2020/06/22 23:59:53 Got message "Hello, world!" upon delivery attempt 824640668296
2020/06/22 23:59:54 Got message "Hello, world!" upon delivery attempt 824637448352
2020/06/22 23:59:55 Got message "Hello, world!" upon delivery attempt 824633943336
2020/06/22 23:59:55 Got message "Hello, world!" upon delivery attempt 824633943448
2020/06/22 23:59:56 Got message "Hello, world!" upon delivery attempt 824633943560
2020/06/22 23:59:57 Got message "Hello, world!" upon delivery attempt 824638259688
2020/06/22 23:59:57 Got message "Hello, world!" upon delivery attempt 824637448752
In other words, the nack'd message is delivered 19 times, nowhere near the 5 times I would expect. If I run the program using the Pub/Sub emulator, I get that the delivery attempt is always 0:
> go run main.go --pubsubEmulatorHost=localhost:8085
2020/06/23 00:00:54 Published message with ID 4
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:56 Got message "Hello, world!" upon delivery attempt 0
...
Here the output is truncated for brevity, but the message is printed about ~200 times (10 times per second for 20 seconds), again far above the 5 times I would expect.
Is the MaxDeliveryAttempts
field of the DeadLetterPolicy
not supposed to limit the number of delivery attempts for nack'd messages? And why is the DeliveryAttempt
field such a weird sequence of integers rather than one that simply increments by 1 each time (cf. https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#Message)?
PUBSUB_SERVICE_ACCOUNT="service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com"
Then:gcloud pubsub topics add-iam-policy-binding <dead letter topic> \ --member="serviceAccount:${PUBSUB_SERVICE_ACCOUNT}"\ --role='roles/pubsub.publisher'
Lastly:gcloud pubsub subscriptions add-iam-policy-binding <subscription with dead letter queue> \ --member="serviceAccount:${PUBSUB_SERVICE_ACCOUNT}"\ --role='roles/pubsub.subscriber'
. Did it work for you? – Alexandre Moraes