Librdkafka: Consumer group support

[vdeters]

Hello, your introduction states that there is no consumer group support, but I want to verify what that means in my context. We have a requirement to run redundant Kafka consumers that consume an identical stream of messages in identical order. For this reason we plan to have these consumers consume from the same topic partition and identify themselves as unique consumer groups. Can I do this using librdkafka? Thanks,
Vince

[edenhill]

Ooh, means I need to update the introduction because there is in fact support for (broker >=0.9) consumer groups!

I dont see any problems with the use-case your are describing.

I will release 0.9.1 shortly, until then I strongly suggest that you use latest master branch.

[vdeters]

Excellent, thanks for your quick response. Follow up question:
These consumers are part of a failover group and need to be aware of each other's presence. I suppose I could use Zookeeper for this purpose, but is there anything already in the protocol that provides awareness of consumer (or producer) groups attached to a specific topic or partition?
Thanks.

Sent from my iPhone

On Apr 6, 2016, at 12:08 PM, Magnus Edenhill [email protected] wrote:

Ooh, means I need to update the introduction because there is in fact support for (broker >=0.9) consumer groups!

I dont see any problems with the use-case your are describing.

I will release 0.9.1 shortly, until then I strongly suggest that you use latest master branch.

—
You are receiving this because you authored the thread.
Reply to this email directly or view it on GitHub

[edenhill]

Awareness is only available within a consumer group, not across them.

[vdeters]

Ok, one more question: for the scenario described in my initial message above, should I use a RdKafka::KafkaConsumer or a RdKafka::Consumer? It doesn't complain if I set "group.id" in a config used by a RdKafka::Consumer, but it's not obvious to me whether the group id is actually sent to the broke in this case. Thanks.

[edenhill]

You should use the KafkaConsumer, it can function both as a balanced consumer group (if you use subscribe()), or as a standalone consumer (if you use assign()).

[jose9r]

Hello, im trying to use KafkaConsumer with one group and with debug options
%7|1460117663.901|TOPIC|rdkafka#consumer-1| New local topic: queue
%7|1460117663.901|DESP|rdkafka#consumer-1| Adding desired topic queue [0]
%7|1460117663.901|FETCHSTART|rdkafka#consumer-1| List with 1 partition(s):
%7|1460117663.901|FETCHSTART|rdkafka#consumer-1| queue [0] offset INVALID
%7|1460117663.901|OFFSET|rdkafka#consumer-1| 10.0.0.203:9092/1: OffsetFetchRequest(v1) for 1/1 partition(s)
%7|1460117663.999|OFFSET|rdkafka#consumer-1| Topic queue [0]: setting default offset INVALID
%7|1460117663.999|OFFSETFETCH|rdkafka#consumer-1| 10.0.0.203:9092/1: OffsetFetchResponse: queue [0] offset -1
%7|1460117663.999|OFFFETCH|rdkafka#consumer-1| 10.0.0.203:9092/1: OffsetFetch for 1/1 partition(s) returned Success
%7|1460117663.999|FETCHSTART|rdkafka#consumer-1| List with 1 partition(s):
%7|1460117663.999|FETCHSTART|rdkafka#consumer-1| queue [0] offset INVALID
%7|1460117663.999|FETCH|rdkafka#consumer-1| Start fetch for queue [0] in state none at offset INVALID (v2)
%7|1460117663.999|PARTSTATE|rdkafka#consumer-1| Partition queue [0] changed fetch state none -> offset-query
%7|1460117663.999|OFFSET|rdkafka#consumer-1| queue [0]: offset reset (at offset INVALID) to END: no previously committed offset available: Local: No offset stored
%7|1460117663.999|OFFSET|rdkafka#consumer-1| queue [0]: no current leader for partition, starting offset query timer for offset END

and i cant read messages, because the offset is setted to the end. When will you probably have available new version with group support??

[jose9r]

Im using kafka_2.11-0.9.0.1 and php-rdkafka from arnaud-lb repository https://github.com/arnaud-lb/php-rdkafka/

[edenhill]

When starting a new consumer group there will be no offsets stored so it will choose the starting offset based on the topic config property auto.offset.reset which defaults to latest (END)

[jose9r]

I set auto.offset.reset to smallest, but it isnt valid to Conf() from KafkaConsumer config.

[edenhill]

That is a topic config property.

Create a topic config object and set auto.offset.reset, then assign the object as the default topic config to your global config object.

[edenhill]

@arnaud-lb Ping ^ not sure if you are exposing the default topic config for high-level consumer users?

[edenhill]

Added to FAQ: https://github.com/edenhill/librdkafka/wiki/FAQ

[arnaud-lb]

@edenhill @jose9r RdKafka\Conf::setDefaultTopicConf() allows to set the default topic config :)

[jose9r]

ok thank you so much!!!

[vdeters]

Can you please provide a cpp KafkaConsumer example that uses assign() instead of subscribe()? rdkafkacpp.h only shows assign() called in the context of a rebalance callback but it's not clear that I even need to implement that callback when I have one consumer per group. It's also not clear to me how the topic is set if subscribe() is not called. Thanks,
Vince

[edenhill]

Balanced high-level consumer

subscribe() takes a absolute list of topics and joins the consumer group configured with group.id with those topics as its subscription. One of the group members will be elected leader and it performs the partition assignment (according partition.assignment.strategy) for all topics subscribed by members of the group.

The rebalance callback is called when a member receives or looses its partition assignment (ASSIGN / REVOKE) and allows an application to perform auxilliery tasks, such as special handling of offset storage. An application must call assign() from within the rebalance to atomically update its partition assignment based on the ASSIGN or REVOKE.

If no rebalance callback is registered librdkafka does the assign() call automatically and no further actions are needed by the application.

Simple consumer

The assign() call may also be used directly with no previous subscribe() call, in this case the consumer will not join any consumer group but immediately start consuming the partitions provided to assign().
To stop consuming simply provide an empty list to assign().

Does that help?

[vdeters]

Thanks that helps somewhat...I thought I could implement multiple simple consumers and call assign() directly but since they don't implement a consumer group they won't be allowed to consume an identical stream from a single partition correct? If so, I need to use balanced high level consumers even though each consumer will be the lone consumer in a unique consumer group, and the rebalance callback is still required. Hope I have that right.

[edenhill]

You should only use the high-level balanced consumer if you want a set of consumers to NOT share messages, i.e., each partition is assigned to only one consumer.
This is what the high-level balanced consumer is all about.

You are saying that you want multiple consumers to consume the same set of partitions, but you also say you want a consumer group. These two statements contradicts since a consumer group will make sure no two consumers share the same partition.

It sounds like you should use assign() directly, and no subscribe().

[vdeters]

If I use assign() directly and assign the same partition to multiple simple consumers, will each consumer receive their own copy of every message in the partition in the same order?

[edenhill]

Yes.

[vdeters]

Ok, so do I even need to set group.id in this case? (I thought consumers needed to identify themselves as unique consumer groups to consume an identical stream from a single partition)

[edenhill]

group.id is needed to find the proper coordinator broker so that offsets can be committed. I suggest you use one group per application.

[vdeters]

(...by single partition, I mean the same partition...)

[jose9r]

so the default option for auto.offset.reset (largest) isnt working for me. Its a general issue or bad configuration? Because when i suscribed to a queue with many messages, i dont want to reprocess old messages.

[edenhill]

@arnaud-lb Will @jose9r need to do anything special with the default config after setting auto.offset.reset? E.g., apply it to the main config or so?

[arnaud-lb]

Yes, the property has to be set before calling setDefaultTopicConf():

$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.offset.reset', 'largest');

$conf->setDefaultTopicConf($topicConf);

$consumer = new RdKafka\KafkaConsumer($conf);

[jose9r]

i have the same config, but when i try to consume always new messages dont appear, and offset is set to end.

[edenhill]

@jose9r Please create a new issue for this problem, and provide output from a run with "debug=topic,fetch" configured.

[vdeters]

So I have 2 simple consumers consuming from the same topic and partition, setting the same group.id, and both setting "auto.offset.reset" = "smallest". They both call assign(). The topic has no messages. When my producer writes to that topic and partition, both consumers get copies of all messages. I assume that's working as designed but if so, I'm a bit confused because it sounds like this case from the kafka doc:

If all the consumer instances have the same consumer group, then this works just like a traditional queue balancing load over the consumers.

...what am I missing? Thanks,
Vince

[edenhill]

The balancing behaviour is only enabled if you use subscribe().
Den 14 apr 2016 15:37 skrev "vdeters" [email protected]:

So I have 2 simple consumers consuming from the same topic and partition,
setting the same group.id, and both setting "auto.offset.reset" =
"smallest". They both call assign(). The topic has no messages. When my
producer writes to that topic and partition, both consumers get copies of
all messages. I assume that's working as designed but if so, I'm a bit
confused because it sounds like this case from the kafka doc:

If all the consumer instances have the same consumer group, then this
works just like a traditional queue balancing load over the consumers.

...what am I missing? Thanks,
Vince

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub
https://github.com/edenhill/librdkafka/issues/601#issuecomment-209945071

[vdeters]

Ok. What are the implications of the consumers using identical vs different group.id values in this case?

[vdeters]

By "this case" I mean the scenario I describe above that uses assign().

[edenhill]

The offset is only commited once (last commit) per topic+partition+groupid, so if you have two consumers in the same group and restart one of them it will restart from the latest commit of any of them, which might not be desirable.

[vdeters]

Ok, understood. But I may want to use identical group ids for another reason, which is the ability of a consumer to detect the presence (and be notified of the disappearance) of other consumers in the group. But this may not be possible using simple consumers - is it even possible using balanced high level consumers?

[edenhill]

You cant really do that with the high level consumer either because only one of the group members are elected group leaders and only the leader will receive the full list of members, the non-leader members will not know about the other members.