重新使用给定时间的 Kafka 消息

Question

我正在使用 Confluence.Kafka .NET 客户端版本 1.3.0。我想从给定时间开始消费消息。

为此，我可以用来OffsetsForTimes获取所需的偏移量以及Commit该分区的偏移量：

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);
}

如果我想跳过消息并跳转到给定的偏移量（如果我想跳转到的偏移量在最后提交的消息之后），则此方法可以正常工作。

但是，如果给定的时间戳早于最后提交的消息的时间戳，则上述方法将不起作用。在上面的代码中，如果在timeStamp最后提交的消息的时间戳之前，那么OffsetsForTimes将返回最后提交的消息的偏移量+1。即使我手动将偏移量设置为较低的偏移量，那么consumer.Commit(newOffsets)似乎没有效果，我我在消费时收到第一条未提交的消息。

有没有办法从代码中实现这一点？

Answer 1

如果您分配给每个分区并指定开始读取的偏移量，则可以做到这一点。

这是获取主题分区列表的方法：

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;
}

这是找到特定时间偏移量的方法：

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();
}

这是通过分配给所有主题分区和特定偏移量来使用的方式：

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();
    }
}

如果您坚持将此应用于消费者组，另一种选择是重置消费者组并使用您的代码，或者创建一个新的消费者组。

Answer 2

我不是专家，但我将尝试向您解释如何做到这一点。

首先，我们必须提到订阅和分配方法。

当您使用订阅时，您传递一个或多个主题。这样，每个主题的分区列表就会根据其组中消费者的数量分配给消费者。主题分区是由主题名称和分区号组成的对象。

consumer.Subscribe(Topic);

您可以使用分配来传递消费者将读取的分区。这个方法不使用消费者的组管理功能（不需要group.id）如果我没记错的话，在分配方法中你可以指定初始偏移量。

consumer.Subscribe(Topic);

另一种选择是使用seek()方法来设置偏移量

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

如果您要混合订阅和分配，请记住您必须先使用取消订阅。

如果您想重新使用所有消息，另一种选择是在新的不同消费者组中创建一个消费者。

示例（供回顾）

我现在就离开你和例子，我稍后会检查。我用java做了这个例子，因为我更熟悉它。在此示例中，我不使用订阅，而是使用分配。检索第一个主题分区，我们设置一个开始日期时间来读取消息，我们创建一个映射，指定每个分区的该日期时间。

通过创建的映射，我们可以使用 offsetsForTimes 方法获取每个分区在指定日期时间的偏移量。根据每个分区的偏移量，我们使用seek 移动到每个分区上的该偏移量，最后我们消费消息。

我现在没有时间检查代码，但我会这样做。我希望它有帮助。

consumer.Seek(topicPartitionOffset);