Confluent-kafka-go: Can't get allow.auto.create.topics=true to work

[tobgu]

Description

After upgrading to v1.6.1 (from v1.1) topics are no longer auto created when subscribing to a topic that does not exist since before.

Previously, as part of test setups for example, I was relying on setting up a subscription and then waiting until a kafka.PartitionEOF was received from consumer.Poll to know that the consumer had received an assignment.

Due to the changes in https://github.com/edenhill/librdkafka/issues/1540 this no longer works which is fine (but a tad backwards incompatible). Instead an error is received, kafka.ErrUnknownTopicOrPart.

I updated the consumer config map to include the new parameter allow.auto.create.topics set to true and expected things to work like before. I did not however, I cannot see any difference whatsoever compared to not specifying the parameter. I still receive the same error with the same error code. When listing the topics from the broker, using ./kafka-topics.sh --bootstrap-server localhost:9092 --list the topic in question does not show up.

It could be that I'm misunderstanding something fundamental here or that I've made a silly mistake somewhere.

How to reproduce

Pseudo code for how to reproduce:

consumer, err := kafka.NewConsumer(&kafkaConfig)
if err != nil {
   panic(err)
}

rebalanceCallback := func(c *kafka.Consumer, event kafka.Event) error {
   log.Printf("Rebalanced: %s", event.String())
   return nil
}

if err := c.consumer.Subscribe(c.topic, rebalanceCallback); err != nil {
   panic(err)
}

for {
    ev := consumer.Poll(c.config.KafkaPollTimeoutMs)
    switch e := ev.(type) {
        case kafka.PartitionEOF:
            // This never happens under v1.16.1, works fine with v1.4.2 for example.
            c.logger.Printf("PartitionEOF: %s", e.String())
        case kafka.Error:
            log.Printf("Kafka error: %s, %s", e.Code(), e.String())
        default:
            c.logger.Printf("Ignored kafka msg: %s\n", e.String())
    }   
}

Checklist

Please provide the following information:

• [X ] confluent-kafka-go and librdkafka version (LibraryVersion(1.6.1 + bundled version of librdkafka)):
• [X] Apache Kafka broker version: 2.3.1
• [X] Client configuration: go kafka.ConfigMap{ "enable.auto.commit": false, "bootstrap.servers": "brokers...", "group.id": "persister", "client.id": "myConsumer", "session.timeout.ms": 6000, "heartbeat.interval.ms": 3000, "broker.address.family": "v4", "auto.offset.reset": "latest", "fetch.wait.max.ms": 2, "fetch.error.backoff.ms": 10, "fetch.min.bytes": 10000, "enable.partition.eof": true, "enable.auto.offset.store": false, "allow.auto.create.topics": true, "topic.metadata.refresh.interval.ms": 1000, }
• [X] Operating system: Linux + MacOS
• [ ] Provide client logs (with "debug": ".." as necessary)
• [ ] Provide broker log excerpts
• [ ] Critical issue

[edenhill]

Subscribed topics will not be auto-created, regardless of allow.auto.create.topics, the topic would be empty anyway.
Topic auto creation really only has value on the producer.

The recommendation is not to rely on auto topic creation at all, but to use the Admin API to create topics.

[tobgu]

OK, thanks. Isn't the documentation for the property terribly misleading then?
https://github.com/edenhill/librdkafka/blame/master/CONFIGURATION.md#L120

It's stated to be a consumer property and the first sentence says: "Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics."

I tried using the admin client as a workaround, which worked fine, but got a bit stuck on #616 which caused some inconvenience.