获取发送到Kafka主题的最后一条消息

Question

我是Kafka的新手，正在开发一个原型，将专有的流媒体服务连接到Kafka中。

我希望得到一个主题上发送的最后一条消息的密钥，因为我们的内部流消费者需要用连接时收到的最后一条消息的ID登录。

我尝试使用使用者执行以下操作，但当同时运行控制台使用者时，我看到消息被重播。

    // Poll so we know we're connected
    consumer.poll(100);
    // Get the assigned partitions
    Set<TopicPartition> assignedPartitions = consumer.assignment();
    // Seek to the end of those partitions
    consumer.seekToEnd(assignedPartitions);

    for(TopicPartition partition : assignedPartitions) {
        final long offset = consumer.committed(partition).offset();
        // Seek to the previous message
        consumer.seek(partition,offset - 1);
    }

    // Now get the last message
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        lastKey = record.key();
    }
    consumer.close();

这是意料之中的行为还是我走错了路？

Answer 1

问题出在final long offset=consumer.committe(partition).offset()上，因为链接api引用committe方法是获取给定分区的最后提交的偏移量，即使用者告诉kafka服务器它已经读取的最后一个偏移量。因此，肯定会重播消息，因为您总是从特定的偏移量读取。因为我想我只需要移除第一个块。