Ability to get consumer group offsets?

[Dave-French]

Hi,

Does anyone know if there are plans to add to the IAdminClient that ability to get consumer group offsets for a given topic? I am assuming that interface would be the logical place for this functionality.

Preferably for the v1 release?

The java client seems to have it available with the following call:
ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId, ListConsumerGroupOffsetsOptions options);

I am using the C# Confluent.Kafka assembly, version 1.0.0-beta2

Thanks,
Dave

[mhowlett]

yes we plan to add this, but it won't make it into 1.0.

[Dave-French]

Excellent, thanks Matt!

By the way, thanks for the great library. It makes it very easy to connect in with C#.

[henrik-mvno]

+1 on this one, would be very usable for us to migrate our microservices onto this client

[tonykimani]

any word on this feature?

[mhowlett]

currently not high on the priority list, thanks for the +1.

[brechtvhb]

+1 would like this feature to integrate within our health checks.

[pitming]

Hi, I'm also very interested. I can try to make the job, but I need a little help on how/where to start. Could you help ?

[PSanetra]

I guess this is blocked by confluentinc/librdkafka#2173

[plachor]

Hi @mhowlett, since there is no built-in solution I tried to add a workaround in my code. According to: confluentinc/librdkafka#2838 (comment) it should be safe to do such without disrupting existing consumer groups:

1. with admin API I fetch existing TopicPartion pairs using GetMetadata
2. I create consumer instance with groupId for which consumer lag I wish to read:

• disabled EnableAutoCommit
• disabled AllowAutoCreateTopics
• disabled EnableAutoOffsetStore
• AutoOffsetReset = AutoOffsetReset.Error

3. I avoid calling subscribe
4. Query committed offsets for selected TopicPartition pairs
5. For each topic for which I find any committed offset I query QueryWatermarkOffsets for each partition in that topic
6. Finally my lag is watermark.High subtract current.Offset

using var adminClient = ..;

// step 1.
var metadata = adminClient.GetMetadata(TimeSpan.FromSeconds(20));
var topicMetadata = metadata.Topics.SelectMany(t => t.Partitions.Select(p => new TopicPartition(t.Topic, p.PartitionId))).ToArray();

// step 2.
using var metadataConsumer = new ConsumerBuilder < byte[], byte[]> (new ConsumerConfig{
        EnableAutoCommit = false,
        AllowAutoCreateTopics = false,
        EnableAutoOffsetStore = false,
        AutoOffsetReset = AutoOffsetReset.Error,
        GroupId = groupId,  // <-- consumer groupId of which lag we wish to check
        BootstrapServers = string.Join(",", settings.BootstrapServers)})
    .SetValueDeserializer(ThrowingDeserializer.Instance)
    .SetKeyDeserializer(ThrowingDeserializer.Instance)
    .SetPartitionsAssignedHandler((Func < IConsumer < byte[], byte[] >, List < TopicPartition >, IEnumerable < TopicPartitionOffset >>)((_, _) => throw new NotSupportedException("Misused.")))
    .Build()

// step 4.
var committedOffsets = metadataConsumer.Committed(topicMetadata, TimeSpan.FromSeconds(10));

// step 5.
var topicsWithFoundOffsets = committedOffsets.GroupBy(t => t.Topic).Where(t => t.Any(s => !s.Offset.IsSpecial)).SelectMany(t => t);

// step 6.
var lag = topicsWithFoundOffsets.Select <TopicPartitionOffset, (TopicPartition, long)>(tpo => {
    if (tpo.Offset.IsSpecial)
        return (tpo.TopicPartition, tpo.Offset);

    var watermark = metadataConsumer.QueryWatermarkOffsets(tpo.TopicPartition, TimeSpan.FromSeconds(10));
    return (tpo.TopicPartition, watermark.High - tpo.Offset);
});

So:
1) Is that snipet safe solution which will not disrupt brokers or consumer groups for which I will check consumer lag?
2) Any suggestions how to improve this workaround ?

[plachor]

Hi @mhowlett, @edenhill could you comment above, am I risking something with this approach when it comes to brokers and performance / stability of consumer groups?

[edenhill]

@plachor That looks like a good solution.

You could even optimize it further by creating the Admin client as a dependent client on top of the Consumer, which will be cheaper and quicker.

Since you're not assigning or subscribing to any partitions you dont need to set these:

        EnableAutoCommit = false,
        AllowAutoCreateTopics = false,
        EnableAutoOffsetStore = false,
        AutoOffsetReset = AutoOffsetReset.Error,

[plachor]

Thanks for review :)

[chenjing1294]

+1 on this one, would be very usable for us to migrate our microservices onto this client