Kafka Consumer lag and batching

[montiver0311]

Description

We have a dotnet kafka consumer running in a continuous web job in Azure. We can't get the consumer to keep up with the amount of messages produced. However, if we create a java client and wire it up we can get the lag to go down and we get caught up to real time. If we switch back on the dot net consumer the message lag starts to increase again.

In java we can configure the consumer to give us a batch of messages and that seems to be the key. Currently we haven't configured fetch max bytes in the dotnet client so it's whatever the default is(52mb i think). Is this possible in the Confluent consumer?

We are running Confluent.Kafka 1.5.0. in a .Net 3.1 worker service in azure app service.

How to reproduce

The vendor kafka is setup so we only get one partition. But this doesn't seem to be an issue for a java client.

Checklist

Please provide the following information:

• A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• Confluent.Kafka nuget version.
• Apache Kafka version.
• Client configuration.
• Operating system.
• Provide logs (with "debug" : "..." as necessary in configuration).
• Provide broker log excerpts.
• Critical issue.

[mhowlett]

the .NET consumer performance is comparable to the java client. but unlike the java client, it delivers messages to your application one at a time from an internal cache. are you calling commit frequently? that would slow things down. if so, consider StoreOffsets.

[montiver0311]

the .NET consumer performance is comparable to the java client. but unlike the java client, it delivers messages to your application one at a time from an internal cache. are you calling commit frequently? that would slow things down. if so, consider StoreOffsets.

It's set to autocommit and the default of 5000 ms.

[mhowlett]

feel free to paste some code

[montiver0311]

//thanks for taking look. it's basic consume example.

public async Task Execute()
{

        try
        {
            using (_consumer)
            {
                _consumer.Subscribe(_streamConsumerSettings.Topic);
               while (!_cancelConsume)
               {                       
                    var result = _consumer.Consume(1000);

                    if (result?.IsPartitionEOF == true)
                    {
                        _logger.LogInformation("IsPartitionEOF = true");
                    }
                    if (result == null)
                    {
                        _logger.LogInformation("result from consume was null.  Continuing");
                        continue;
                      
                    }
                    var topicPartitionOffset = result?.TopicPartitionOffset;
                    var equipmentEx = _deserializer.Deserialize(result.Message.Value);

                    var mappedEquipment = new Equipment();

                    try
                    {
                        mappedEquipment = _mapper.Map<Equipment>(equipmentEx);
                        mappedEquipment.TopicParitionOffset = topicPartitionOffset?.ToString();

                         await _eventHubClient.SendAsync(mappedEquipment);
                       
                    }
                catch (OutOfMemoryException me)
                {
                    _logger.LogError(me, me.StackTrace ?? me.Message);
                    
                }
                catch (Exception ex)
                {
                    _logger.LogInformation("Bad mapping");
                    _logger.LogError(ex, ex.StackTrace ?? ex.Message);
                    
                }
              }
            }
        }
        catch (ConsumeException e)
        {
            _logger.LogError(e, e.StackTrace ?? e.Message);
            _cancelConsume = true;
            _consumer.Close();
            RestartConsumer();
        }
    }

//here's my config object.
private static ConsumerConfig GetConfig()
{
var config = new ConsumerConfig
{
BootstrapServers = _streamConsumerSettings.Bootstrapservers,
GroupId = _streamConsumerSettings.GroupId,
AutoOffsetReset = AutoOffsetReset.Latest,
AutoCommitIntervalMs = 20000,
SaslUsername = _streamConsumerSettings.Username,
SaslPassword = _streamConsumerSettings.Password,
SecurityProtocol = SecurityProtocol.SaslSsl,
SaslMechanism = SaslMechanism.Plain,
LogConnectionClose = true,
FetchMaxBytes = 100000000,
FetchMinBytes = 52000000,
EnableAutoCommit = true
};
return config;
}

[montiver0311]

I updated a few values as I was chatting. so those are just local changes and I don't have the fetch min/max running on the real web job. Same with the AutoCommitInterval. That's default.

[montiver0311]

[2020-09-23 16:13:00,741] INFO [GroupCoordinator 1]: Assignment received from leader for group testclient for generation 2120 (kafka.coordinator.group.GroupCoordinator)
[2020-09-23 16:22:50,424] INFO [GroupCoordinator 1]: Member rdkafka-f0e673f3-2cfd-4d4b-be68-b34fdc857ac7 in group testclient has failed, removing it from the group

[mhowlett]

probably best not to mix async and sync code (the consume method is sync). just using .Result should be ok, but this shouldn't be causing an issue. looks ok. enable debugging set Debug config property to all for more information about why the consumer is being kicked out of the group.

[montiver0311]

%7|1600890755.843|RECV|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Received JoinGroupResponse (v5, 209 bytes, CorrId 89, rtt 3381.78ms)
%7|1600890755.843|JOINGROUP|rdkafka#consumer-1| [thrd:main]: JoinGroup response: GenerationId 2160, Protocol range, LeaderId rdkafka-c6e0aac7-646f-4b26-9296-96725de53de8 (me), my MemberId rdkafka-c6e0aac7-646f-4b26-9296-96725de53de8, 1 members in group: (no error)
%7|1600890755.845|JOINGROUP|rdkafka#consumer-1| [thrd:main]: Elected leader for group "testclient" with 1 member(s)
%7|1600890755.846|GRPLEADER|rdkafka#consumer-1| [thrd:main]: Group "testclient": resetting group leader info: JoinGroup response clean-up
%7|1600890755.846|CGRPJOINSTATE|rdkafka#consumer-1| [thrd:main]: Group "testclient" changed join state wait-join -> wait-metadata (v21, state up)
%7|1600890755.846|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Request metadata for 1 topic(s): partition assignor
%7|1600890755.847|SEND|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Sent MetadataRequest (v4, 60 bytes @ 0, CorrId 90)
info: StreamConsumer[0]
result from consume was null. Continuing
%7|1600890756.016|RECV|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Received MetadataResponse (v4, 282 bytes, CorrId 90, rtt 168.65ms)
%7|1600890756.017|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: ===== Received metadata (for 1 requested topics): partition assignor =====
%7|1600890756.018|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: ClusterId: nefLlgJkQ06m_yUnsyj8yQ, ControllerId: 2
%7|1600890756.018|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: 3 brokers, 1 topics
%7|1600890756.018|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Broker #0/3: streamin-url:31090 NodeId 0
%7|1600890756.018|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Broker #1/3: streamin-url:31092 NodeId 2
%7|1600890756.019|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Broker #2/3: streamin-url:31091 NodeId 1
%7|1600890756.020|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Topic #0/1: topic-name with 1 partitions
%7|1600890756.021|METADATA|rdkafka#consumer-1| [thrd:main]: Topic topic-name partition 0 Leader 1
%7|1600890756.022|METADATA|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: 1/1 requested topic(s) seen in metadata
%7|1600890756.022|ASSIGN|rdkafka#consumer-1| [thrd:main]: Group "testclient" running range assignment for 1 member(s):
%7|1600890756.023|ASSIGN|rdkafka#consumer-1| [thrd:main]: Member "rdkafka-c6e0aac7-646f-4b26-9296-96725de53de8" (me) with 1 subscription(s):
%7|1600890756.024|ASSIGN|rdkafka#consumer-1| [thrd:main]: topic-name [-1]
%7|1600890756.025|ASSIGN|rdkafka#consumer-1| [thrd:main]: range: Topic topic-name with 1 partition(s) and 1 subscribing member(s)
%7|1600890756.026|ASSIGN|rdkafka#consumer-1| [thrd:main]: range: Member "rdkafka-c6e0aac7-646f-4b26-9296-96725de53de8": assigned topic topic-name partitions 0..0
%7|1600890756.027|ASSIGN|rdkafka#consumer-1| [thrd:main]: Group "testclient" range assignment for 1 member(s) finished in 4.254ms:
%7|1600890756.041|ASSIGN|rdkafka#consumer-1| [thrd:main]: Member "rdkafka-c6e0aac7-646f-4b26-9296-96725de53de8" (me) assigned 1 partition(s):
%7|1600890756.042|ASSIGN|rdkafka#consumer-1| [thrd:main]: topic-name [0]
%7|1600890756.043|ASSIGNOR|rdkafka#consumer-1| [thrd:main]: Group "testclient": "range" assignor run for 1 member(s)
%7|1600890756.044|CGRPJOINSTATE|rdkafka#consumer-1| [thrd:main]: Group "testclient" changed join state wait-metadata -> wait-sync (v21, state up)
%7|1600890756.045|SEND|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Sent SyncGroupRequest (v3, 200 bytes @ 0, CorrId 91)
%7|1600890756.045|BROADCAST|rdkafka#consumer-1| [thrd:GroupCoordinator]: Broadcasting state change
%7|1600890756.428|RECV|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Received SyncGroupResponse (v3, 62 bytes, CorrId 91, rtt 383.07ms)
%7|1600890756.429|SYNCGROUP|rdkafka#consumer-1| [thrd:main]: SyncGroup response: Success (52 bytes of MemberState data)
%7|1600890756.441|ASSIGN|rdkafka#consumer-1| [thrd:main]: Group "testclient": new assignment of 1 partition(s) in join state wait-sync
%7|1600890756.442|DESIRED|rdkafka#consumer-1| [thrd:main]: topic-name [0]: marking as DESIRED
%7|1600890756.443|BARRIER|rdkafka#consumer-1| [thrd:main]: Group "testclient": rd_kafka_cgrp_assign:2600: new version barrier v22
%7|1600890756.444|ASSIGN|rdkafka#consumer-1| [thrd:main]: Group "testclient": assigning 1 partition(s) in join state wait-sync
%7|1600890756.445|CGRPJOINSTATE|rdkafka#consumer-1| [thrd:main]: Group "testclient" changed join state wait-sync -> assigned (v22, state up)
%7|1600890756.446|BARRIER|rdkafka#consumer-1| [thrd:main]: Group "testclient": rd_kafka_cgrp_partitions_fetch_start0:1878: new version barrier v23
%7|1600890756.447|FETCHSTART|rdkafka#consumer-1| [thrd:main]: Group "testclient": starting fetchers for 1 assigned partition(s) in join-state assigned (usable_offsets=no, v23, line 2651)
%7|1600890756.447|FETCHSTART|rdkafka#consumer-1| [thrd:main]: List with 1 partition(s):
%7|1600890756.471|FETCHSTART|rdkafka#consumer-1| [thrd:main]: topic-name [0] offset INVALID
%7|1600890756.479|OFFSET|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: OffsetFetchRequest(v1) for 1/1 partition(s)
%7|1600890756.480|OFFSET|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Fetch committed offsets for 1/1 partition(s)
%7|1600890756.482|SEND|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Sent OffsetFetchRequest (v1, 88 bytes @ 0, CorrId 92)
%7|1600890756.482|HEARTBEAT|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: Heartbeat for group "testclient" generation id 2160
%7|1600890756.507|SEND|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Sent HeartbeatRequest (v3, 94 bytes @ 0, CorrId 93)
%7|1600890756.653|RECV|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Received OffsetFetchResponse (v1, 58 bytes, CorrId 92, rtt 170.27ms)
%7|1600890756.653|OFFSETFETCH|rdkafka#consumer-1| [thrd:main]: List with 1 partition(s):
%7|1600890756.663|OFFSETFETCH|rdkafka#consumer-1| [thrd:main]: topic-name [0] offset INVALID
%7|1600890756.664|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic topic-name [0]: setting default offset INVALID
%7|1600890756.665|OFFSETFETCH|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: OffsetFetchResponse: topic-name [0] offset 186450163, metadata 0 byte(s)
%7|1600890756.665|OFFFETCH|rdkafka#consumer-1| [thrd:main]: GroupCoordinator/1: OffsetFetch for 1/1 partition(s) returned Success
%7|1600890756.666|BARRIER|rdkafka#consumer-1| [thrd:main]: Group "testclient": rd_kafka_cgrp_partitions_fetch_start0:1878: new version barrier v24
%7|1600890756.667|FETCHSTART|rdkafka#consumer-1| [thrd:main]: Group "testclient": starting fetchers for 1 assigned partition(s) in join-state assigned (usable_offsets=yes, v24, line 1810)
%7|1600890756.668|FETCHSTART|rdkafka#consumer-1| [thrd:main]: List with 1 partition(s):
%7|1600890756.669|FETCHSTART|rdkafka#consumer-1| [thrd:main]: topic-name [0] offset 186450163
%7|1600890756.669|CGRPJOINSTATE|rdkafka#consumer-1| [thrd:main]: Group "testclient" changed join state assigned -> started (v24, state up)
%7|1600890756.689|BARRIER|rdkafka#consumer-1| [thrd:main]: topic-name [0]: rd_kafka_toppar_op_fetch_start:2299: new version barrier v22
%7|1600890756.690|CONSUMER|rdkafka#consumer-1| [thrd:main]: Start consuming topic-name [0] at offset 186450163 (v22)
%7|1600890756.706|OP|rdkafka#consumer-1| [thrd:main]: topic-name [0] received op FETCH_START (v22) in fetch-state stopped (opv21)
%7|1600890756.753|FETCH|rdkafka#consumer-1| [thrd:main]: Start fetch for topic-name [0] in state stopped at offset 186450163 (v22)
%7|1600890756.788|PARTSTATE|rdkafka#consumer-1| [thrd:main]: Partition topic-name [0] changed fetch state stopped -> active
%7|1600890756.791|FETCH|rdkafka#consumer-1| [thrd:main]: Partition topic-name [0] start fetching at offset 186450163
%7|1600890756.807|WAKEUP|rdkafka#consumer-1| [thrd:main]: sasl_ssl://streamin-url:31091/1: Wake-up
%7|1600890756.823|RECV|rdkafka#consumer-1| [thrd:GroupCoordinator]: GroupCoordinator/1: Received HeartbeatResponse (v3, 6 bytes, CorrId 93, rtt 315.34ms)
%7|1600890756.830|FETCHDEC|rdkafka#consumer-1| [thrd:sasl_ssl://streamin-url:31091/boot]: Topic topic-name [0]: fetch decide: updating to version 22 (was 18) at offset 186450163 (was 186512161)
%7|1600890756.830|CGRPOP|rdkafka#consumer-1| [thrd:main]: Group "testclient" received op PARTITION_JOIN in state up (join state started, v24) for topic-name [0]
%7|1600890756.912|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://streamin-url:31091/boot]: sasl_ssl://streamin-url:31091/1: Topic topic-name [0] in state active at offset 186450163 (0/100000 msgs, 0/65536 kb queued, opv 22) is fetchable
%7|1600890756.924|PARTADD|rdkafka#consumer-1| [thrd:main]: Group "testclient": add topic-name [0]
%7|1600890756.948|FETCHADD|rdkafka#consumer-1| [thrd:sasl_ssl://streamin-url:31091/boot]: sasl_ssl://streamin-url:31091/1: Added topic-name [0] to fetch list (1 entries, opv 22, 0 messages queued): fetchable
%7|1600890756.964|FETCH|rdkafka#consumer-1| [thrd:sasl_ssl://streamin-url:31091/boot]: sasl_ssl://url-obfuscated

[mhowlett]

that looks fine. how many messages per second are you getting and how large are the messages?

[montiver0311]

here's the message size is a byte[139]. i'm trying to get conduktor setup so i can see some more information about the lag.

[mhowlett]

await _eventHubClient.SendAsync(mappedEquipment); is synchronous, so it'll be slow

[montiver0311]

that could be the problem too eventually. this is the first issue. var equipmentEx = _deserializer.Deserialize(result.Message.Value); Hand rolled deserialization of the Avro file. When i take this out I can see the conduktor lag go way down. Thanks for the help! I just need to get these other pieces out of this loop.

[neerajk22]

Same case with me also getting performance issue while using below consumer.

In our use case we want to show some charts, metrices and grid based on Kafka topics data.( All Topics are already loaded with Json data from different systems )
We are planning to use Kafka connect and will sync topics data to Cassandra database.
Based on some trigger like any new data in Kafka topic will re-load UI and read same data from Cassandra (Via Dot Net core APIs) and display it on UI.
So is it good idea to use Kafka connect and sync data to Cassandra and query on Cassandra to load UI data Realtime.

Note : Reading data directly from Kafka topics and display on UI using Dot net Kafka consumer is very slow as in our use case we need to query 4-5 topics to get required data.

private List<Dictionary<string, object>> RetrieveDataWithGenericRecordKey(CachedSchemaRegistryClient schemaRegistryClient, string topicName, CancellationTokenSource cts)
{
var processedResult = new List<Dictionary<string, object>>();

        using (var consumer = new ConsumerBuilder<GenericRecord, GenericRecord>(consumerConfig)
            .SetKeyDeserializer(new CustomAvroDeserializer<GenericRecord>(schemaRegistryClient).CustomAsSyncOverAsync())
            .SetValueDeserializer(new CustomAvroDeserializer<GenericRecord>(schemaRegistryClient).CustomAsSyncOverAsync())
            .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
            .Build())
        {

            Dictionary<string, object> results = new Dictionary<string, object>();

            consumer.Subscribe(topicName);

            try
            {
                while (true)
                {
                    try
                    {
                        // consumer.Subscribe(topicName);

                        ConsumeResult<GenericRecord, GenericRecord> consumeResult = consumer.Consume(cts.Token);

                        if (consumeResult.IsPartitionEOF)
                        {
                            return processedResult;
                        }

                        Dictionary<string, object> obj = new Dictionary<string, object>();
                        var result = ProcessRetrievedResult(consumeResult);
                        processedResult.Add(result);

                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Error occurred: {e}");

                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Ensure the consumer leaves the group cleanly and final offsets are committed.
                consumer.Close();
            }
        }

        return processedResult;
    }