Logstash 5.1.1 kafka 输入不会获取有关主题的现有消息

Question

我有以下带有 kafka 输入的 logstash 配置

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["mytopic"]
  }
}
filter {
  json {
    source => "message"
  }
}
output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "my_index"
    codec => "json"
    document_id => "%{id}"
    doc_as_upsert => true
    action => "update"
  }
}

我面临的问题是，当我运行 logstash 时，它不会拾取有关该主题的旧消息。我的印象是，第一次运行 logstash 时，它会拾取有关主题的所有尚未使用的消息。我检查了这是一个新主题，并且其中有一些消息，当它开始运行时，logstash 没有接收到这些消息。它确实会在运行时接收有关该主题的消息，但不会接收在其开始之前存在的消息。我是在配置中遗漏了什么，还是输入本身的怪癖。消息的保证对我的业务需求至关重要。

Answer 1

由于您没有为 kafka 指定组 id，因此重要的注意事项如下：

• Kafka group.id（logstash kafka 配置中的 group_id）设置为 logstash 的默认值，即“logstash”
• logstash 中 enable.auto.commit (enable_auto_commit) 的默认 Kafka 值为“true”
• Kafka auto.offset.reset (auto_offset_reset) 在 logstash 中没有默认值，所以我假设使用的是 Kafka 默认值 latest。

因此，当您在某个主题上运行消费者并且它无法获取该主题中已有的消息时，可能会发生以下两种情况之一：

1. 不存在与消费者具有相同组 ID 的现有组，因此使用 Kafka 默认 auto.offset.reset 值 latest 并且消费者将忽略已经存在的消息。
2. 存在具有相同组 ID（“logstash”）的现有组，并且具有该组 ID 的某些使用者已经使用了现有消息并提交了偏移量（该其他使用者可能是您之前运行的或某些其他具有相同组 id 的消费者）。这意味着该组下的其他消费者不会重新使用这些消息，除非以某种方式明确告知这样做。

所以你可能想要做的是设置一些 Kafka 配置，对于 logstash 你应该能够设置

group_id => "some_random_group"

auto_offset_reset => "最早的"

如果你现在运行消费者，由于 some_random_group 没有现有的偏移量并且重置是最早的，消费者应该消费一个主题中的所有现有消息并提交偏移量。这意味着如果在消费完所有消息后再次运行消费者，它不会消费现有的消息。

Answer 2

您应该将 kafka 输入插件设置 auto_offset_reset 设置为“最早”。

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    auto_offset_reset => "earliest"
    topics => ["mytopic"]
  }
}