Librdkafka: Improvement: KafkaConsumer without group

[DEvil0000]

I the most cases it makes sense to use the _group.id_ for consumer group (partition assignment) and offset storage.
We also have cases where we do not want to store the offset but have a consumer group. This should be possible by setting _enable.auto.commit=false_.
But it seams like there are also cases where we do not want stored offsets and partition assignment at all. The new KafkaConsumer forces me to insert a _group.id_. This kind of makes sense in the normal cases but feels strange in this case. (If I do not set it I get _"group.id" must be configured_)
I know two workarounds for this:

1. set a unique group.id
2. use the old consumer class.

The second option feels wrong because the handling of both APIs is different. I would need to maintain both wrappings and such.
The first option feels better but is still a hack.

How about allowing an empty _group.id_ or adding a configuration key for _enable.partition.assignment_?

[edenhill]

Cant you just use a special group.id that you know no managed group will use? e.g., "my_unmanaged_group"
You can share this group among all unmanaged clients.

[DEvil0000]

Then they would still do partition assignment right?
But even if I would do this and write a special kind of rebalance_cb 'hack' this would just be another workaround for something like _group.id=""_ or _enable.partition.assignment=false, enable.auto.commit=false_.

[edenhill]

Partition assignment is only performed if you use subscribe(), if you use assign() then it will be skipped.
There is no rebalance_cb when just using assign().

[DEvil0000]

using assign:
can I use TopicPartition with partition -1 for all?
otherwise I need to know the partition count.

can I call it from outside rebalance_cb() without limitations? (multi thread/timing/locking/conditions)
might bring a lot of frustration and strange behaviour.

and I need to keep the TopicPartition objects and maintain their live cycle - right?

[edenhill]

1: No, you will need to know the number of partitions first (query by metadata()).
2: Yes, you can call it from outside rebalance_cb.

[DEvil0000]

Seams like this is again a pretty ugly workaround. I prefer the unique group id workaround.
As the title says this is no bug or completely missing thing.
This is a suggested improvement to get better usability.

[edenhill]

Absolutely, I hear you, I will mark it as an enhancement for future releases.

[eugpermar]

This issue is interesting for me :) I currently use a unique consumer group id for every consumer I launch to achieve it. Would it have any drawbacks?

I would like also to know if there is a way to always start consuming from offset 0 if I use subscribe().
I'm trying to use kafka log compaction and I need to consume all messages from beginning, but even if I disable commit.auto.store and offset.store.method in my configuration, it always start consuming from last offset of every partition.

Thanks in advance!

[ottomata]

+1.

I'm building a websockets <-> Kafka interface just for consuming, that will be exposed to the internet. I don't want to have Kafka managing client info for the internet at large. assign() will work, but having to manually specify the partitions means that running consumers won't notice if we add partitions to their assigned topics, right?

Aside from that one rare use case, metadata() + assign() I think will work.

[edenhill]

Narrow use-case, big testing effort.

[DEvil0000]

@eugpermar sorry for the late reply (have missed this somehow) but you should use:

• unique group.id
• enable.auto.commit=false
• auto.offset.reset=smallest (or earliest/beginning)
• (and maybe make sure there was no offset for this group.id)

@ottomata you should use the high level KafkaConsumer, periodically update metadata or build a proxy abstraction to handle all this for you. Btw there is a rest proxy out there.

[pipefail]

Not sure if this factors into the decision here regarding the wontfix label-- but the Java API explicitly supports opting in and out of Kafka's consumer group management features if you want to manage your own partition assignment strategies if they diverge from the built-in failure detection / rebalancing logic.

While I can see the large testing effort, I'm not sure I'd agree that the use case(s) is necessarily "narrow". A case I've seen more than once now is when you want multiple consumers instances as "standby" processors for each partition and you want to manage fail-over / leader change via an alternate consensus system (outside of Kafka/ZK).

Sure, you can accomplish this by hacking the group.id or launching multiple consumers in the same process with different group.id's, but you are still forced to cause Kafka to run the overhead of it's rebalancing procedure which introduces latency and a mandatory synchronization barrier that might not be applicable to an application architecture that is managing it's own failure detection protocols. Causing Kafka to spawn a bunch of one-off CG generations for each consumer instance also just feels wasteful and is an obvious circumvention of the feature.

I know this is an old thread and not likely to be revisited, but given my experience with the Java API and this exact use case of specifically opting out of rebalancing, I think there is a lot of value here.

[edenhill]

@petermelias How is this done in the Java consumer?

[pipefail]

@edenhill Instead of setting group ID, calling subscribe and implementing rebalance handler, you just use the assign API directly, from the Javadocs:

It is also possible for the consumer to manually assign specific partitions (similar to the older "simple" consumer) using assign(Collection). In this case, dynamic partition assignment and consumer group coordination will be disabled.

Details here: https://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

I've spent a bit more time familiarizing myself with librdkafka since my previous post and it looks like this same effect is possible using the legacy consumer API? I think maybe the distinction is that this behavior/mode of operation is not considered deprecated/legacy in Javaland whereas with librdkafka it seems the high-level consumer implementation mandates the use of a group and is the preferred approach now?

Given your comment above about the rebalance_cb being disabled when using assign in librdkafka that feels consistent with the Java API-- perhaps the only difference is the requirement of the group id when using the new consumer? The Java API allows the new consumer class to be used _without_ specifying a group.id so long as you don't try to use subscribe.

Feels like it would be conceptually equivalent then to use the old consumer (and use assign) here in librdkafka but then I wonder if there are any hidden drawbacks to that given its status as legacy? Or, as you've suggested, use the new consumer, give a dummy group.id and just never call subscribe?

[edenhill]

From a broker perspective the group.id is needed to commit offsets (since they are keyed on topic,partition,groupid).

Internally a group.id is required to fire up the new consumer mode, but if the consumer doesn't subscribe() or commit then the group.id is not used in any communication with the cluster and can be set to anything.

This means you can indeed use assign() as with the Java client.

[pipefail]

Awesome, thank you.