Sarama: [Question] How to Gracefully shutdown Kafka consumer process

[alok87]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.

| Sarama | Kafka | Go |
|--------|-------|----|
| master | 2.5.0 | 1.14.6 |

How to gracefully shutdown the processMessage() on OS signals.

func (c consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
        for message := range claim.Messages() {
        err := c.processMessage(session, message)
        }
}

Requirement

the processMessage() batches messages in memory and on reaching 100, it does some processing, uploads to s3 and then do a commit.

Problem

I want to shutdown this processMessage() gracefully by listening to OS signals in every loop of processing of message. But since the context signals are not passed to the ConsumeClaim(), I am not able to forward them to the processMessage() routine.

Is not it possible to have context ctx in ConsumeClaim() defination ?

consumerGroupSession has ctx. But the idiomatic way in Go is to pass ctx (context) in functions and not use it from struct{}.
Please let me know your thoughts on this.

Thank you for open sourcing the library 👍

[d1egoaz]

I don't _think_ you need to check the OS signals on the processing of the messages.

I _think_ you can simply defer the consumer group close function and make the OS signals handler end the execution of the main go routine.

something like:

func yourMainMethod() {
...
    group, err := sarama.NewConsumerGroup...
    if err != nil {
        logrus.WithError(err).Error("Couldn't create a Kafka consumer")
        return
    }

    defer func() {
        err = group.Close()
        if err != nil {
            logrus.WithError(err).Error("Error closing consumer group")
            return
        }
    }()

signal handling here

}

signal handling can be similar to https://github.com/Shopify/sarama/blob/master/examples/consumergroup/main.go#L116-L126

having ctx on our methods would be nice though, we will need new methods to not break users' code

[d1egoaz]

since the context signals are not passed to the ConsumeClaim(), I am not able to forward them to the processMessage()

you could also add a channel to your consumer and send the event when the signal handling is triggered.

[alok87]

Thanks for the suggestions. @d1egoaz (i will explore them)
I was thinking of using Context() in the session since c.processMessage(session, message) already has the session object.

Something like:

func (c consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
        for message := range claim.Messages() {
        err := c.processMessage(session, message)
        }
}

func processMessage(session, message) error {
      // make memoryMessages until the batch threshold reach

    for id, data := range memoryMessages {
        select {
        default:
            message := data.(*sarama.ConsumerMessage)
            b.handleMessage(message, id)
        case <-b.session.ctx.Done():
            return nil
        }
    }
    return nil
}

Thoughts? Does it look good?

[d1egoaz]

@alok87 right, yes, looks like a good idea!,so you can break the bulk processing if the consumer is going down! @bai @dnwe any more comments? 👀

it also seems you can pass your own Context when calling group.Consume(ctx,....

[alok87]

We should definitely have an example for this in examples. Since had to add context handling here (and many other places also). I am thinking there could be a better way to do this.

    for message := range claim.Messages() {
    select {
    case <-session.Context().Done():
        klog.Infof("Gracefully shutdown. Stopped taking new messages.")
        return nil
        default:
        err := c.processMessage(session, message)
        if err != nil {
            klog.Errorf("Error processing: value:%s, timestamp:%v, topic:%s", string(message.Value), message.Timestamp, message.Topic)
        continue
        }
    }
    }

[d1egoaz]

@alok87 PRs Welcomes :D

it should be a great addition to our examples 🚀

[alok87]

Sure I will add that may be we can take the discussion of what would be better way to handle signal shutdown there...