Use a message for a topic in Confluent.Kafka when consumer run

Viewed 628

I'm using Confluent.Kafka(1.4.4) in a .netCore project as a message broker. In the startup of the project I set only "bootstrapservers" to the specific servers which were in the appSetting.json file and I produce messages in an API when necessary with the code below in related class:

public async Task WriteMessage<T>(string topicName, T message)
    {
        using (var p = new ProducerBuilder<Null, string>(_producerConfig).Build())
        {
            try
            {
                var serializedMessage= JsonConvert.SerializeObject(message);
                var dr = await p.ProduceAsync(topicName, new Message<Null, string> { Value = serializedMessage });
                logger.LogInformation($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
            }
            catch (ProduceException<Null, string> e)
            {
                logger.LogInformation($"Delivery failed: {e.Error.Reason}");
            }
        }
    }

I have also added the following code In the consumer solution :

public async Task Run()
    {
        using (var consumerBuilder = new ConsumerBuilder<Ignore, string>(_consumerConfig).Build())
        {
            consumerBuilder.Subscribe(new List<string>() { "ActiveMemberCardForPanClubEvent", "CreatePanClubEvent", "RemovePanClubEvent"
            });

            CancellationTokenSource cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true; // prevent the process from terminating.
                cts.Cancel();
            };

            try
            {
                while (true)
                {
                    try
                    {
                        var consumer = consumerBuilder.Consume(cts.Token);
                        if (consumer.Message != null)
                        {
                            using (LogContext.PushProperty("RequestId", Guid.NewGuid()))
                            {
                                //Do something
                                logger.LogInformation($"Consumed message '{consumer.Message.Value}' at: '{consumer.TopicPartitionOffset}'.");
                                await DoJob(consumer.Topic, consumer.Message.Value);
                                consumer.Topic.Remove(0, consumer.Topic.Length);
                            }

                        }
                        else
                        {
                            logger.LogInformation($"message is null for topic '{consumer.Topic}'and partition : '{consumer.TopicPartitionOffset}' .");
                            consumer.Topic.Remove(0, consumer.Topic.Length);
                        }
                    }
                    catch (ConsumeException e)
                    {

                        logger.LogInformation($"Error occurred: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Ensure the consumer leaves the group cleanly and final offsets are committed.
                consumerBuilder.Close();
            }
        }
    }

I produce a message and when the consumer project is run everything goes perfectly and the message is being read in the consumer solution. The problem is raised when the consumer project is not run and I queue a message in the API with the message producer in API. After running consumers there is not any valid message for that topic that it's message is being produced. I am familiar with and have experiences with message brokers and I know that by sending a message it will be on the bus until it is being used but I don't understand why it doesn't work with Kafka in this project.

1 Answers

The default setting for the "auto.offset.reset" Consumer property is "latest".

That means (in the context of no offsets being written yet) if you write a message to some topic and then subsequently start the consumer, it will skip past any messages written before the consumer was started. This could be why your consumer is not seeing the messages queued by your producer.

The solution is to set "auto.offset.reset" to "earliest" which means that the consumer will start from the earliest offset on the topic.

https://docs.confluent.io/current/installation/configuration/consumer-configs.html#auto.offset.reset

Related