Kafkaのオフセット指定で読み込む

2年 ago
蓮, 翼
1 minute

Kafkaは、メッセージを処理する際に、特定の位置にあるメッセージを読み取るためにオフセットを指定することができます。以下は、オフセットを使用してメッセージを読み取るための手順です。

  1. Kafkaコンシューマー
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  1. assign()
TopicPartition topicPartition = new TopicPartition("my-topic", 0);
consumer.assign(Collections.singletonList(topicPartition));
consumer.seek(topicPartition, desiredOffset);
  1. メッセージを消費し始めます。
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
    consumer.commitSync();
}

Assignでは消費するトピックとパーティションを指定し、Seekでは開始するオフセットを指定します。pollでメッセージを取得し、CommitSyncで消費したオフセットを手動コミットします。desiredOffsetには、どのオフセットからメッセージを読み始めるのかを代入します。

オフセット指定読み込みの場合、指定したオフセットが有効（対応するトピックおよびパーティション内に存在する）であることをあらかじめ確認してください。そうしないと、メッセージが読み取れない、または予期しないメッセージが読み取られる可能性があります。

bannerAds