Sarama: add consumers for consumer group

[bluesky1024]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.
Sarama Version:2.2.0
Kafka Version:2.2.0
Go Version:1.11.5

Configuration

What configuration values are you using for Sarama and Kafka?
kafka:
broker:1
topic: "my-topic"-3partition-1 replicas

the sarama producer:

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewHashPartitioner
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    config.Version = sarama.V2_2_0_0

    producer, err := sarama.NewSyncProducer([]string{"0.0.0.0:9092"}, config)

Logs

When filing an issue please provide logs from Sarama and Kafka if at all
possible. You can set sarama.Logger to a log.Logger to capture Sarama debug
output.

Problem Description

according to the example given in https://github.com/Shopify/sarama/tree/master/examples/consumergroup
the consumergroup is built, and the consumer is defined as follows

        ...
    ctx := context.Background()
    client, err := sarama.NewConsumerGroup(strings.Split(brokers, ","), group, config)
    if err != nil {
        panic(err)
    }

    consumerA := Consumer{
        ready:     make(chan bool, 0),
    }

    go func() {
        for {
            err := client.Consume(ctx, strings.Split(topics, ","), &consumerA)
            if err != nil {
                panic(err)
            }
        }
    }()

    <-consumerA.ready // Await till the consumer has been set up
    log.Println("Sarama consumerA up and running!...")

However, when i want to add more consumer in the same consumerGroup client or just go run one more the same file, panic appears. Is there more example about how to add more consumer in the same consumerGroup?
thanks for your attention.

[junweiyang]

i have the same issue。

[varun06]

hmm, I will have a look this weekend.

[varun06]

Alright, I ran same example locally and I was able to run more than 1 consumer per consumer group. it didn't panic at all and it subscribed to partitions in topic fine.

I have a 10 partition topic btw.

That's the command I ran locally multiple time to have multiple consumers.

go run main.go -brokers localhost:9092 -group test -topics consumer-group-test -verbose

@bluesky1024 ^^

[bluesky1024]

@varun06 Thanks for your relay. I can not figure out what‘s wrong with my code...
I tried to run the same file. The result is as follows
a.

2019/04/28 13:15:37 Starting a new Sarama consumer
[sarama] 2019/04/28 13:15:37 Initializing new client
[sarama] 2019/04/28 13:15:37 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:37 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:37 client/metadata fetching metadata for all topics from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:15:37 Connected to broker at 127.0.0.1:9092 (unregistered)
[sarama] 2019/04/28 13:15:37 client/brokers registered new broker #1001 at 127.0.0.1:9092
[sarama] 2019/04/28 13:15:37 Successfully initialized new client
[sarama] 2019/04/28 13:15:37 client/metadata fetching metadata for [my-topic] from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:15:37 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:15:37 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
[sarama] 2019/04/28 13:15:37 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:37 Connected to broker at 127.0.0.1:9092 (registered as #1001)
[sarama] 2019/04/28 13:15:37 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:15:37 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
2019/04/28 13:15:37 Sarama consumerA up and running!...
[sarama] 2019/04/28 13:15:37 consumer/broker/1001 added subscription to my-topic/2
[sarama] 2019/04/28 13:15:37 consumer/broker/1001 added subscription to my-topic/0
[sarama] 2019/04/28 13:15:37 consumer/broker/1001 added subscription to my-topic/1
[sarama] 2019/04/28 13:15:55 consumer/broker/1001 closed dead subscription to my-topic/2
[sarama] 2019/04/28 13:15:55 consumer/broker/1001 closed dead subscription to my-topic/0
[sarama] 2019/04/28 13:15:55 consumer/broker/1001 closed dead subscription to my-topic/1
[sarama] 2019/04/28 13:15:55 client/metadata fetching metadata for [my-topic] from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:15:55 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:15:55 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
panic: close of closed channel

b.

2019/04/28 13:15:53 Starting a new Sarama consumer
[sarama] 2019/04/28 13:15:53 Initializing new client
[sarama] 2019/04/28 13:15:53 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:53 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:53 client/metadata fetching metadata for all topics from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:15:53 Connected to broker at 127.0.0.1:9092 (unregistered)
[sarama] 2019/04/28 13:15:53 client/brokers registered new broker #1001 at 127.0.0.1:9092
[sarama] 2019/04/28 13:15:53 Successfully initialized new client
[sarama] 2019/04/28 13:15:53 client/metadata fetching metadata for [my-topic] from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:15:53 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:15:53 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
[sarama] 2019/04/28 13:15:53 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2019/04/28 13:15:53 Connected to broker at 127.0.0.1:9092 (registered as #1001)
[sarama] 2019/04/28 13:15:55 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:15:55 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
2019/04/28 13:15:55 Sarama consumerA up and running!...
[sarama] 2019/04/28 13:15:55 consumer/broker/1001 added subscription to my-topic/2
[sarama] 2019/04/28 13:16:08 consumer/broker/1001 closed dead subscription to my-topic/2
[sarama] 2019/04/28 13:16:08 client/metadata fetching metadata for [my-topic] from broker 127.0.0.1:9092
[sarama] 2019/04/28 13:16:08 client/coordinator requesting coordinator for consumergroup example1 from 127.0.0.1:9092
[sarama] 2019/04/28 13:16:08 client/coordinator coordinator for consumergroup example1 is #1001 (127.0.0.1:9092)
panic: close of closed channel

it seems that the first process has added subscription of my-topic 's 3 partition. When the second process added subscription of the same topic's same partition, panic appear, which causing dead subscription.

Is it possible that there is something wrong on my configure of kafka?

[varun06]

Can you please share kafka configs?

[bluesky1024]

@varun06 I'm terribly sorry for forgetting check my email these days...
I use the docker images from https://github.com/wurstmeister/kafka-docker. And the YML configure file is as follow

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    depends_on: [ zookeeper ]
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 192.168.1.9
      KAFKA_CREATE_TOPICS: "my-topic:3:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

[bluesky1024]

@hilalevavi I got the same problem as yours. Here is my program, is there something wrong?
https://github.com/bluesky1024/goMblog/blob/20190425_mblog_to_feed/daemon/kafka/relation/main.go

[prune998]

Did you set a different config.ClientID in each run ? (not sure if it makes a difference)

[bluesky1024]

@prune998 Thanks for your advice very much. In fact, my program run normally now. However, I do not figure out why my process failed before...😂
And it proves that the config.ClientId is defined as string "sarama" commonly if not specific this parameter. These processes run normally without set different clientId.

[d1egoaz]

If I understood correctly, is this issue already solved @bluesky1024 ?

can we close the issue?

[bluesky1024]

If I understood correctly, is this issue already solved @bluesky1024 ?

can we close the issue?

yes, thanks