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?