PubSub isn't acknowledging messages

Question

I have a pubsub subscription (all default settings except the number of go-routines is 1000), and for some reason messages never get acknowledged, and therefore redelivered. Redelivery is taking between 1 and 2 minutes. I'm calling message.Ack() less than 1 second after the message is received, so I don't understand what is happening. It shouldn't be because of latency between the app and pubsub itself, because after publishing a message to the topic, the message is delivered practically immediately.

The subscription has an acknowledgement deadline of 10 seconds. I tried increasing this to 120, but the same problem still occurred. I can't think of any reason why these messages aren't being acknowledged, and therefore being redelivered.

Code for reference:

if err := pubsubSubscription(client).Receive(ctx, func(lctx context.Context, message *pubsub.Message) {
    log.Println("Received message") // occurs < 1s after publishing
    ack := message.Ack  
    if err := adapters.Handle(conn, id, gatewayAddr, message.Data); err != nil {
        log.Println("Will nack message")
        ack = message.Nack // not reached (in this context/example)
        cancel()
    }
    log.Println("Will ack message") // occurs ~200µs after message receipt
    ack()
}); err != nil {
    return fmt.Errorf("unable to subscribe to PubSub messages: %s", err)
}

To clarify, I've only published 1 message to the topic, but that callback is called every 1 or 2 minutes infinitely.

EDIT

This only occurs when the number of go-routines in the subscription receive settings is set to a number higher than runtime.NumCPU(). Is this the expected behaviour? If so, how does this work with Kubernetes (which I'm using)?

EDIT 2 -- request for full code for reproduction

const (
    DefaultMaxOutstandingMessages = 1000000
    DefaultMaxOutstandingBytes    = 1e9
)

func SubscribeToTables(id int) error {
    var opts []option.ClientOption
    if sa := os.Getenv("SERVICE_ACCOUNT"); sa != "" {
        opts = append(opts, option.WithCredentialsJSON([]byte(sa)))
    }

    ctx := context.Background()
    projectID := os.Getenv("PROJECT_ID")
    client, err := pubsub.NewClient(ctx, projectID, opts...)
    if err != nil {
        return fmt.Errorf("error creating GCP PubSub client: %s", err)
    }

    cctx, cancel := context.WithCancel(ctx)
    go func() {
        qch := make(chan os.Signal)
        signal.Notify(qch, os.Interrupt, syscall.SIGTERM)
        <-qch
        cancel()
    }()

    mch := make(chan *pubsub.Message)
    gatewayAddr := os.Getenv("GATEWAY_ADDRESS")
    conn, err := adapters.GetGatewayConn(gatewayAddr)
    if err != nil {
        return fmt.Errorf("unable to connect to Gateway: %s", err)
    }
    go func() {
        for {
            select {
            case message := <-mch:
                if err := adapters.Handle(conn, id, gatewayAddr, message.Data); err != nil {
                    cancel()
                    return
                }
                message.Ack()
            case <-ctx.Done():
                return
            }
        }
    }()
    if err := pubsubSubscription(client).Receive(cctx, func(_ context.Context, message *pubsub.Message) {
        mch <- message
    }); err != nil {
        return fmt.Errorf("unable to subscribe to PubSub messages: %s", err)
    }
    return nil
}

func pubsubSubscription(client *pubsub.Client) *pubsub.Subscription {
    sub := client.Subscription(os.Getenv("SUBSCRIPTION_ID"))
    sub.ReceiveSettings = pubsub.ReceiveSettings{
        MaxExtension:       pubsub.DefaultReceiveSettings.MaxExtension,
        MaxExtensionPeriod: pubsub.DefaultReceiveSettings.MaxExtensionPeriod,
        MaxOutstandingMessages: parsePubSubReceiveSetting(
            "MAX_OUTSTANDING_MESSAGES",
            "max outstanding messages",
            DefaultMaxOutstandingMessages,
        ),
        MaxOutstandingBytes: parsePubSubReceiveSetting(
            "MAX_OUTSTANDING_BYTES",
            "max outstanding bytes",
            DefaultMaxOutstandingBytes,
        ),
        NumGoroutines: parsePubSubReceiveSetting( // if this is higher than runtimie.NumCPU(), the aforementioned issue occurs 
            "NUM_GO_ROUTINES",
            "Go-routines",
            1000, 
        ),
    }
    return sub
}

func parsePubSubReceiveSetting(env, name string, defaultValue int) int {
    e := os.Getenv(env)
    i, err := strconv.Atoi(e)
    if err != nil {
        log.Printf("Unable to parse number of GCP PubSub %s. Can't parse '%s' as int", name, e)
        log.Printf("Using default number of %s (%d)", name, defaultValue)
        return defaultValue
    }
    return i
}

guillaume blaquiere guillaume blaquiere · Accepted Answer · 2020-04-24T19:46:29

I suspect that you exit too quickly your code. You have to cancel() the context for stopping the Receive loop and flushing the data back to PubSub.

Try to add cancel() just after your ack()

PubSub isn't acknowledging messages

1 Answers