Sarama: getting i/o timeout error when using sarama consumer group

Created on 15 Oct 2018  路  2Comments  路  Source: Shopify/sarama

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.
Sarama Version: v1.19.0
Kafka Version: v2.0
Go Version: v1.11

Problem Description

Getting i/o timeout error from broker when using consumer group.

Error:          Received unexpected error read tcp <ip>:63133-><ip>:9092: i/o timeout

Error is returned from https://github.com/Shopify/sarama/blob/master/broker.go#L592

Note: same kafka cluster is working fine when not using consumer group

Most helpful comment

I experienced perhaps the same problem. One consumer instance using the ConsumerGroup API is fine, but when I tried to start another instance, the other instance got the same error and could not join the group.

Sarama v1.19.0
Kafka 1.1.0
Go v1.10.3

Solved the problem by watching ConsumerGroupSession.Context().Done() as well in ConsumeClaim.
The comment at https://github.com/Shopify/sarama/blob/v1.19.0/consumer_group.go#L18 is clear enough, but it seems that the example at https://godoc.org/github.com/Shopify/sarama#example-ConsumerGroup does not take rebalancing into consideration.

func (h MyHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    var msg SyncMessage
    for {
        select {
        case cMsg := <-claim.Messages():
            err := json.Unmarshal(cMsg.Value, &msg)
            if err != nil {
                return err
            }
            // do something
            sess.MarkMessage(cMsg, "")
        case <-sess.Context().Done():
            return nil
        }
    }
}

All 2 comments

We're experiencing the same problem and cranked up the time outs. It did not resolve the issue. It looks like a connectivity issue, though we are experiencing the same problem on serval deployments which excludes that this is hardware related. We are getting stuck in a loop of rebalances.

Sarama v1.19.0
Sarama Cluster v2.1.15
Kafka v2.0.
Go v1.11

Reverting Sarama v1.17 did not help.

I experienced perhaps the same problem. One consumer instance using the ConsumerGroup API is fine, but when I tried to start another instance, the other instance got the same error and could not join the group.

Sarama v1.19.0
Kafka 1.1.0
Go v1.10.3

Solved the problem by watching ConsumerGroupSession.Context().Done() as well in ConsumeClaim.
The comment at https://github.com/Shopify/sarama/blob/v1.19.0/consumer_group.go#L18 is clear enough, but it seems that the example at https://godoc.org/github.com/Shopify/sarama#example-ConsumerGroup does not take rebalancing into consideration.

func (h MyHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    var msg SyncMessage
    for {
        select {
        case cMsg := <-claim.Messages():
            err := json.Unmarshal(cMsg.Value, &msg)
            if err != nil {
                return err
            }
            // do something
            sess.MarkMessage(cMsg, "")
        case <-sess.Context().Done():
            return nil
        }
    }
}
Was this page helpful?
0 / 5 - 0 ratings

Related issues

nicklipple picture nicklipple  路  4Comments

qiuyesuifeng picture qiuyesuifeng  路  3Comments

blackfox1983 picture blackfox1983  路  3Comments

damiannolan picture damiannolan  路  4Comments

damiannolan picture damiannolan  路  7Comments