Cannot re-process old Kafka messages because of Quarkus overriding the group.id

[nicolaferraro]

I'm running a simple integration in Camel K:

from('kafka:bitcoin-kafka-topic?brokers=my-cluster-kafka-bootstrap.kafka-demo.svc.cluster.local:9092&groupId=myowngroupid&autoOffsetReset=earliest')
    .to('log:info')

I'm using myowngroupid to reprocess old messages using a different consumer group. But when I run this integration, I notice that the Kafka group Id is always a fixed string camel-k-integration.

The reason seems to be due to this part of the code:

camel-quarkus/extensions/kafka/runtime/src/main/java/org/apache/camel/quarkus/component/kafka/QuarkusKafkaClientFactory.java

Lines 63 to 67 in 9344d52

	if (quarkusKafkaConfiguration != null) {
	for (Map.Entry<String, Object> entry : quarkusKafkaConfiguration.entrySet()) {
	camelKafkaProperties.put(entry.getKey(), entry.getValue());
	}
	}

Which seems correct at first glance, but unfortunately the Quarkus config unexpectedly contains wrong values at runtime:

[oscerd]

I think this behavior should be configurable. If I don't want to merge configurations or I don't care about the quarkus extension workflow, I should be able to avoid it.

[jamesnetherton]

Sorry about this one, it's my bad. The map put should of course be putIfAbsent. I think we can make the properties merge behaviour configurable based on whether the Quarkus k8s service binding capability is present. And if it is, only then do we set up the custom QuarkusKafkaClientFactory.

I'll also look into tweaking the Quarkus Kafka extension, as I don't see the point of it adding the group.id config, even when no other Kafka configuration has been provided.

[oscerd]

I don't if there are other extension approaching the configuration in this way, but making it configurable would be really nice. At least on the kamelet side, we will be sure what we are configuring will be effectively used.

[oscerd]

Thanks for looking into this.

[jamesnetherton]

I created a PR that should fix this #2913. To my knowledge, none of the other Camel projects dependent on CQ is explicitly forcing usage of quarkus-kubernetes-service-binding, so having a check for whether the capability is present and not setting up the custom KafkaClientFactory should be good enough.

I need to do some follow up research for changing the map put to putIfAbsent. Turns out it's a bit tricky to have this because of the defaults that the Camel component sets for various options. However, I introduced a new config option to disable the config merging if required.

[cescoffier]

For the record, we merged a fix in Quarkus.