Method ConsumeClaim

examples/consumer_load_aware/main.go:130–145

(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim)

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error { return nil }

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case message, ok := <-claim.Messages():
			if !ok {
				return nil
			}
			c.inFlight.Add(1)
			log.Printf("Message claimed: topic=%s partition=%d offset=%d", message.Topic, message.Partition, message.Offset)
			session.MarkMessage(message, "")
			c.inFlight.Add(-1)
		case <-session.Context().Done():
			return nil
		}
	}
}