Consumer.SubscribeTopics for more than one topic is not working about confluent-kafka-go

[qrpike]

@edenhill #115 turned out to be my issue. Working fine now.

[edenhill]

Do both topics exist in the cluster?

[felipejfc]

yes, they do, I've also tested it and reproduced locally...

[felipejfc]

@edenhill I've tested now using a regex and it works (only 1 topic passed to SubscribeTopics containing a regex that matches two topics) but still, sending two topics in a slice makes it only subscribe to the last one

[edenhill]

I can't reproduce this:

$ ./consumer_example :9093 g1 xtest1 xtest2 xtest3
Created Consumer rdkafka#consumer-1
% Message on xtest1[0]@0:
msg1
% Reached EOF at xtest1[0]@1(Broker: No more messages)
% Message on xtest2[0]@0:
msg2
% Message on xtest3[0]@0:
msg3
% Reached EOF at xtest2[0]@1(Broker: No more messages)
% Reached EOF at xtest3[0]@1(Broker: No more messages)
Ignored OffsetsCommitted (<nil>, [xtest1[0]@1 xtest2[0]@1 xtest3[0]@1])

Is your q.assignPartitions() loosing the extra topics perhaps?

[edenhill]

Can you print q.Topics prior to calling SubscribeTopics?

[felipejfc]

@edenhill is it possible that my assignPartitions is overwriting assignments?

func (q *KafkaConsumer) assignPartitions(partitions []kafka.TopicPartition) error {
	l := q.Logger.WithFields(logrus.Fields{
		"method":     "assignPartitions",
		"partitions": fmt.Sprintf("%v", partitions),
	})

	l.Debug("Assigning partitions...")
	err := q.Consumer.Assign(partitions)
	if err != nil {
		l.WithError(err).Error("Failed to assign partitions.")
		return err
	}
	l.Info("Partitions assigned.")
	return nil
}

full code is here

https://github.com/topfreegames/pusher/blob/master/extensions/kafka_consumer.go

if you do not find something that may be causing this, I will investigate it further, for now I got it to work using regex anyway, so this is not critic for me anymore, but I'm willing to help if you need

[edenhill]

The code looks fine, can't see anything obvious.
Could you add some debugging and see where the topic lists gets lost?

[qrpike]

Did anyone solve this issue?

Thanks,

[edenhill]

@qrpike What is the exact issue you are seeing? And do you have some code to show?