[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...

[alok87]

Hey recenty I obeserved an issue in which the context I passed to sarama is active. But sarama session context session.Context(). gets closed.

When can sarama session context get closed?

i think session gets closed and its context closed when the ConsumerClaim() func gets completed for any of the topic in the consumer group. If that is the case I can not use session's Context for batch go routines shutdown. As a single empty topic would cause repeated shutdown.

Please suggest an alternate way.

[alok87]

Please help out here. Passing context as parameter is the right way based on Golang documentation. golang/go#22602

I have to go around this by putting context in the struct.

[alok87]

Closed in favour of #1892