Hono: AbstractAtLeastOnceKafkaConsumer may consume nothing when a leader for a subscribed topic is not elected yet

[fkaltner]

AbstractAtLeastOnceKafkaConsumer first calls subscribe and then uses poll to query records from Kafka.

The call to subscribe though does not mean necessarily mean that the assignment of a consumer to a partition was made. This can be seen whenever the consumer subscribes to a topic which does not yet exist and right afterwards a record is published to this topic auto-creating it. Hono's integration tests do follow this pattern: first the northbound consumer subscribes and right afterwards a message is published by a protocol adapter.

Consequently this leads to a race condition: if the assignment of consumers to partitions could not be done before the first poll is triggered then the client polls without a partition assignment, i.e. it polls nothing which each poll. This does not lead to an exception being thrown (afaik the Apache Kafka Client itself does not throw an error in that case as well), the client just does not receive any messages.

The described behaviour can be observed by the following logs (look for the line The following subscribed topics are not assigned to any members):

13:56:00.110 [vert.x-kafka-consumer-thread-1] WARN  o.apache.kafka.clients.NetworkClient - [Consumer clientId=telemetry-e4e91f55-f62b-4709-a21e-9c61bd5ece34, groupId=its-805d4582-3c66-49c5-bbe0-665ac4755f20] Error while fetching metadata with correlation id 2 : {hono.telemetry.d89cefce-3641-48d7-8664-08e85ff49226=LEADER_NOT_AVAILABLE}
13:56:00.110 [vert.x-kafka-consumer-thread-0] WARN  o.apache.kafka.clients.NetworkClient - [Consumer clientId=event-598c3d47-0859-4b2c-af74-d55ba22eb05e, groupId=its-805d4582-3c66-49c5-bbe0-665ac4755f20] Error while fetching metadata with correlation id 2 : {hono.event.d89cefce-3641-48d7-8664-08e85ff49226=LEADER_NOT_AVAILABLE}
13:56:00.151 [vert.x-kafka-consumer-thread-1] WARN  o.apache.kafka.clients.NetworkClient - [Consumer clientId=telemetry-e4e91f55-f62b-4709-a21e-9c61bd5ece34, groupId=its-805d4582-3c66-49c5-bbe0-665ac4755f20] Error while fetching metadata with correlation id 7 : {hono.telemetry.d89cefce-3641-48d7-8664-08e85ff49226=LEADER_NOT_AVAILABLE, hono.event.d89cefce-3641-48d7-8664-08e85ff49226=LEADER_NOT_AVAILABLE}
13:56:00.153 [vert.x-kafka-consumer-thread-1] WARN  o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=telemetry-e4e91f55-f62b-4709-a21e-9c61bd5ece34, groupId=its-805d4582-3c66-49c5-bbe0-665ac4755f20] The following subscribed topics are not assigned to any members: [hono.telemetry.d89cefce-3641-48d7-8664-08e85ff49226, hono.event.d89cefce-3641-48d7-8664-08e85ff49226]

I did a bit of research but I could not find a workaround yet. An obvious solution would be to use the partitionsAssignedHandler of Vert.x's KafkaConsumer, so that the poll will only be started whenever the consumer was properly assigned to all partitions. However this partitionsAssignedHandler is never called unless a poll is triggered (or a handler was set on the consumer, which then leads to losing control over the poll loop). So it feels like a hen and egg problem to me.

[kaniyan]

It makes sense to wait until the partitions have been assigned. The same scenario has been handled by making use of the partitionsAssignedHandler(...) in KafkaBasedInternalCommandConsumer and KafkaBasedCommandConsumerFactoryImpl.

[b-abel]

The same scenario has been handled by making use of the partitionsAssignedHandler(...) in KafkaBasedInternalCommandConsumer and KafkaBasedCommandConsumerFactoryImpl.

@kaniyan Have you tried to combine this will manual offset commit?

[kaniyan]

The same scenario has been handled by making use of the partitionsAssignedHandler(...) in KafkaBasedInternalCommandConsumer and KafkaBasedCommandConsumerFactoryImpl.

@kaniyan Have you tried to combine this will manual offset commit?

@b-abel Currently the above mentioned classes don't employ manual offset commit. I didn't get it why manual offset commit is relevant in the context of this issue. My suggestion is that in AbstractAtLeastOnceKafkaConsumer, once we subscribe to a _topic/topic pattern_, to wait until the partitions are assigned before calling kafkaConsumer.poll(...).

[fkaltner]

@kaniyan The problem is that a call to subscribe only does not lead to any partition assignment as far as I observed it. That's what I wanted to say with the last section:

However this handler is never called unless a poll is triggered (or a handler was set on the consumer, which then leads to losing control over the poll loop).

Or did I miss something here and this is indeed possible?

[b-abel]

I didn't get it why manual offset commit is relevant in the context of this issue.

Without manual offset commit you have zero delivery guarantees, messages can be delivered multiple times or never. FMPOV this is completely useless for messaging in Hono.

[kaniyan]

I didn't get it why manual offset commit is relevant in the context of this issue.

Without manual offset commit you have zero delivery guarantees, messages can be delivered multiple times or never. FMPOV this is completely useless for messaging in Hono.

I am aware of the manual offset commit and zero delivery guarantees and hence there is a TODO in the above mentioned classes (KafkaBasedInternalCommandConsumer & KafkaBasedCommandConsumerFactoryImpl) to incorporate it. I didn't get the relevance of your comment regarding manual offset commit with regard to this issue. I understood now that the intention of the comment was to find out if manual offset commits were incorporated in the above classes.

[fkaltner]

Together with @calohmn I investigated this further. A possible (working 😃 ) option would be to use partitionsFor to query the partitions for the topics right after the call to subscribe.

If done this way, I couldn't reproduce the error on my machine - seems like the call to partitionsFor is only answered when all consumers are assigned to partitions. In the callback the call to poll can be made then:

public Future<Void> start() {

        final Promise<Void> promise = Promise.promise();
        if (topics != null) {
            kafkaConsumer.subscribe(topics, promise);
        } else {
            kafkaConsumer.subscribe(topicPattern, promise);
        }

        final Promise<KafkaConsumerRecords<String, Buffer>> pollPromise = Promise.promise();
        kafkaConsumer.partitionsFor(topics.iterator().next(), partitions -> {
            System.out.printf("GOT %s PARTITIONS FOR %s", partitions.result().size(), topics.iterator().next());

            pollPromise.future()
                    .onSuccess(this::handleBatch) // do not wait for the processing to finish
                    .recover(cause -> Future.failedFuture(new KafkaConsumerPollException(cause)));
            kafkaConsumer.poll(pollTimeout, pollPromise);
        });

        return CompositeFuture.all(promise.future(), pollPromise.future()).mapEmpty();
}

Of course this call would be needed for all partitions and the topic pattern...

[kaniyan]

@kaniyan The problem is that a call to subscribe only does not lead to any partition assignment as far as I observed it. #

@fkaltner After _subscribe_, I noticed that the _partitionsAssignedHandler_ is being invoked. My understanding is that this handler is invoked as the partitions have been assigned. After taking a closer look with debug statements, even though _partitionsAssignedHandler_ has been invoked but still no partitions have been assigned. I get it what you meant by handler is never called.

[fkaltner]

@kaniyan Just to make sure we're on the same page:

to my understanding the partitionsAssignedHandler is only invoked when you called subscribe and set a handler in KafkaConsumer by calling handler (see: https://vertx.io/docs/vertx-kafka-client/java/#_receiving_messages_from_a_topic_joining_a_consumer_group)

or

after calling subscribe and poll (see https://vertx.io/docs/vertx-kafka-client/java/#_receiving_messages_with_explicit_polling).

I didn't observe the case with the _partitionsAssignedHandler_ being invoked but no partitions being assigned as far as I can remember. But, good to know, so one would need to check for the number of assigned partitions.

[b-abel]

I understood now that the intention of the comment was to find out if manual offset commits were incorporated in the above classes.

@kaniyan What I am interested in is if it is possible to use the abstraction the Vert.x KafkaProducer provides combined with manual offset committing. My understanding was that you can either use the handler and the helping handlers like partitionsAssignedHandler with automatic offset commit (or maybe external offset storage). _Or_ you do not use all of that and have to use the explicit polling and commiting, i.e. control the loop yourself.
If it is true that we cannot mix, we need to avoid the abstractions of the KafkaConsumer.
The background of my question was, that I wanted to know if you tested if this might be false and we could use the handlers of the abstraction and combine this with explicit committing. Because that would be an interesting option.

[b-abel]

@fkaltner Do you know if this problem occurs only with automatic topic creation? If so, it might not be worth a lot of effort, because I think we should expect that the external topics (event, telemetry, C&C) are created upfront. I would not recommend using automatic topic creation there.

[fkaltner]

Do you know if this problem occurs only with automatic topic creation? If so, it might not be worth a lot of effort, because I think we should expect that the external topics (event, telemetry, C&C) are created upfront. I would not recommend using automatic topic creation there.

@b-abel To my current understanding of the problem this is true. I also talked briefly with @calohmn about the topics better being created upfront in my opinion (as we do currently for AMQP as well) but we didn't come to a conclusion about this. From your comment I get that this is still to be discussed?

[fkaltner]

Attaching a log file of a failed test run illustrating the problem with Kafka Client logs on verbose level.

test.log

[fkaltner]

Thinking about up-front topic creation:

I guess, from a DevOps perspective, this makes sense. Usually the number of partitions is limited in a Kafka cluster. So I think the right point in time to fail would be when a new tenant is created and the resources in a cluster are not sufficient anymore. If this is not done when the tenant is created, this could lead to lots of tenants being created and then, when first messages arrive or subscriptions are made, that the topics for those tenant cannot be created completely (they may have only a event or telemetry topic then for instance).

[b-abel]

As far as I know, automatic topic creation is not recommended for production setups. For Hono, the idea was that when a tenant is created, the tenant manager also creates its topics and corresponding ACLs.
Automatic topic creation is configured at the broker. So we cannot assume that a cluster has this enabled. It looks like Confluent disables it too: https://riferrei.com/2020/03/17/why-the-property-auto-create-topics-enable-is-disabled-in-confluent-cloud/

[fkaltner]

@b-abel Is this already implemented for any of Hono's registries? If not, shall we create an issue for that or are you already on it?

@calohmn & @kaniyan What about C&C? Do you need automatic topic creation for command routing?

[calohmn]

I agree: I think explicit topic creation when provisioning the tenant should be the norm and would prevent possible timeouts when accessing the topics for the first time (because of the time it takes to create the topic and propagate the metadata).
And yes, I guess in many environments, auto.create.topics.enable would be false anyway.

Nonetheless I think we should support the case that the topics might get auto-created on first access, as such a Kafka configuration could be common in testing/evaluation scenarios and would be rather convenient.

In any case, I think it still needs to be investigated what the real issue behind the The following subscribed topics are not assigned to any members error here is. So I would focus on that here first.

[kaniyan]

Whenever a tenant is being registered, then the necessary topics and ACLs should also be added. I am also of the same opinion that the topics are not be created on the fly by enabling auto.create.topics.enable to _true_.

[fkaltner]

It turned out that setting the Kafka client's configuration property auto.offset.reset(see https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_auto.offset.reset) to earliest makes the problem (consumer is not assigned to any partition) disappear on my machine (which is macOS based) as well on a Linux machine running the same test in a Jenkins job.

I cannot make any sense of that - at the time of this writing I assume it is more a side-effect of setting this property.

(when writing integration tests I became aware that not all expected messages were received which turned out to be due the default value auto.offset.reset. In the process I noticed that this also make the aforementioned "workaround" meaning using partitionsFor obsolete).