Re-Consume Kafka messages from a given time

Viewed 5404

I am using Confluent.Kafka .NET client version 1.3.0. I would like to start consuming messages from a given time onwards.

To do so, I could use OffsetsForTimes to get the desired offset and Commit that offset for that partition:

private void SetOffset()
{
    const string Topic = "myTopic";
    const string BootstrapServers = "server1, server2";

    var adminClient = new AdminClientBuilder(
        new Dictionary<string, string>
        {
            { "bootstrap.servers", BootstrapServers },
            { "security.protocol", "sasl_plaintext" },
            { "sasl.mechanisms", "PLAIN" },
            { "sasl.username", this.kafkaUsername },
            { "sasl.password", this.kafkaPassword }
        }).Build();

    var consumer = new ConsumerBuilder<byte[], byte[]>(
        new Dictionary<string, string>
        {
            { "bootstrap.servers", BootstrapServers },
            { "group.id", this.groupId },
            { "enable.auto.commit", "false" },
            { "security.protocol", "sasl_plaintext" },
            { "sasl.mechanisms", "PLAIN" },
            { "sasl.username", this.kafkaUsername },
            { "sasl.password", this.kafkaPassword }
        }).Build();

    // Timestamp to which the offset should be set to
    var timeStamp = new DateTime(2020, 3, 1, 0, 0, 0, DateTimeKind.Utc);

    var newOffsets = new List<TopicPartitionOffset>();
    var metadata = adminClient.GetMetadata(Topic, TimeSpan.FromSeconds(30));
    foreach (var topicMetadata in metadata.Topics)
    {
        if (topicMetadata.Topic == Topic)
        {
            foreach (var partitionMetadata in topicMetadata.Partitions.OrderBy(p => p.PartitionId))
            {
                var topicPartition = new TopicPartition(topicMetadata.Topic, partitionMetadata.PartitionId);

                IEnumerable<TopicPartitionOffset> found = consumer.OffsetsForTimes(
                    new[] { new TopicPartitionTimestamp(topicPartition, new Timestamp(timeStamp, TimestampType.CreateTime)) },
                    TimeSpan.FromSeconds(5));

                newOffsets.Add(new TopicPartitionOffset(topicPartition, new Offset(found.First().Offset)));
            }
        }
    }

    consumer.Commit(newOffsets);

    // Consume messages
    consumer.Subscribe(Topic);
    var consumerResult = consumer.Consume();
    // process message
    //consumer.Commit(consumerResult);
}

This works fine if I want to skip messages and jump to a given offset if the offset to which I would like to jump is after the last committed message.

However, the above approach won't work if the given timestamp is before the timestamp of the last committed message. In the above code, if the timeStamp is before the timestamp of the last committed message, then OffsetsForTimes will return the offset of that last committed message + 1. Even if I manually set the offset to a lower offset, then consumer.Commit(newOffsets) seems to have no effect and I am getting the first uncommitted message when consuming.

Is there a way to achive this from the code?

2 Answers

You can do it if you assign to each partition and state the offset from which to start reading.

this is how you get the list of topic partitions:

public static List<TopicPartition> GetTopicPartitions(string bootstrapServers, string topicValue) {
    var tp = new List<TopicPartition>();
    using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build()) {
        var meta = adminClient.GetMetadata(TimeSpan.FromSeconds(20));
        meta.Topics.ForEach(topic => {
            if (topic.Topic == topicValue) {
                foreach (PartitionMetadata partition in topic.Partitions) {
                    tp.Add(new TopicPartition(topic.Topic, partition.PartitionId));
                }
            }
        });
    }
    return tp;
}

This is how you find the offset of a particular time:

List<TopicPartition> topic_partitions = frmMain.GetTopicPartitions(mBootstrapServers, txtTopic.Text);

using (var consumer = new ConsumerBuilder<Ignore, string>(cfg).Build()) {
    consumer.Assign(topic_partitions);

    List<TopicPartitionTimestamp> new_times = new List<TopicPartitionTimestamp>();
    foreach (TopicPartition tp in topic_partitions) {
        new_times.Add(new TopicPartitionTimestamp(tp, new Timestamp(dtpNewTime.Value)));
    }

    List<TopicPartitionOffset> seeked_offsets = consumer.OffsetsForTimes(new_times, TimeSpan.FromSeconds(40));
    string s = "";
    foreach (TopicPartitionOffset tpo in seeked_offsets) {
        s += $"{tpo.TopicPartition}: {tpo.Offset.Value}\n";
    }
    Console.WriteLine(s);
    consumer.Close();
}

This is how you consume by assigning to all topic partition and specific offsets:

using (var consumer =
    new ConsumerBuilder<string, string>(config)
        .SetErrorHandler((_, e) => Log($"Error: {e.Reason}"))
        .Build()) {
    consumer.Assign(seeked_offsets);

    try {
        while (true) {
            try {
                var r = consumer.Consume(cancellationToken);
                // do something with r
            } catch (ConsumeException e) {
                //Log($"Consume error: {e.Error.Reason}");
            }
        }
    } catch (OperationCanceledException) {
        //Log("Closing consumer.");
        consumer.Close();
    }
}

The other option, if you insist in applying this to the consumer group, would be to reset the consumer group and use your code, or to create a new consumer group.

I'm not an expert but i'm going to try to explain how you could do it.

In first place, we have to mention subscribe and assign methods.

When you use subscribe, you pass a one or more topics. With this, a list of partitions of each topics are assigned to the consumer depending of the number of consumers in its group. A topic partition is an object formed by the topic name and the partition number.

consumer.Subscribe(Topic);

You can use assign to pass the partitions of which the consumer will read. This method does not use the consumer's group management functionality (where no need of group.id) If i'm not wrong, in assign method you can specify the initial offset.

consumer.Assign(topicName, 0, new Offset(lastConsumedOffset));
consumer.Assign(topicPartition, new Offset(lastConsumedOffset));

Another option is to use seek() method to set the offset

consumer.Seek(topicPartitionOffset);

If you are going to mix subscribe and assign remember that you have to use unsubscribe before.

Another option, if you want to re-consume all the messages is create a consumer in a new different consumer group.

EXAMPLE (TO REVIEW)

I'm leaving you and example for now, i will check it later. I have done the example in java because i'm more familiar with it. In this example, i don't use subscribe, i use assign. First topic partitions are retrieved, we set a start datetime to read messages from, we create a map specifing that that datetime for each partition.

With the created map we get the offset of each partition on the specified datetime with offsetsForTimes method. With the offset of each partition we use seek to move to that offset on each partitions and finally we consume the messages.

I don't have time now to check the code, but i will do it. I hope it helps.

        AdminClient client = AdminClient.create(getAdminClientProperties());
        KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<String, GenericRecord>(
                getConsumerProperties());

        String TOPIC = "topic";

        // get info of all partitions of a topic
        List<PartitionInfo> partitionsInfo = consumer.partitionsFor(TOPIC);

        // create TopicPartition list
        Set<TopicPartition> partitions = new HashSet<>();
        for (PartitionInfo p : partitionsInfo) {
            partitions.add(new TopicPartition(p.topic(), p.partition()));
        }

        // Consumer will read from all partitions
        consumer.assign(partitions);
        DateTime timeToStartReadMessagesFrom = new DateTime(2020, 3, 1, 0, 0, 0);

        Map<TopicPartition, Long> timestamps = new HashMap<>();
        for (TopicPartition tp : partitions) {
            timestamps.put(tp, timeToStartReadMessagesFrom.getMillis());
        }
        // get the offset for that time in each partition
        Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestamps);
        for (TopicPartition tp : partitions) {
            consumer.seek(tp, offsets.get(tp).offset());
        }

        while (true) {
            final ConsumerRecords<String, GenericRecord> consumerRecords = consumer.poll(1000);
            // do something
            break;
        }
        consumer.close();
        System.out.println("DONE");
Related