ウェブサイト検索

CLIを通じてデータを受信するためのKafkaコンシューマーの設定方法

紹介

Apache Kafkaは、Kafkaクラスタへの基本的なテキストメッセージの生成と消費のためのシェルスクリプトを提供します。これらは探索や実験には便利ですが、実際のアプリケーションはプログラム的にKafkaにアクセスします。この目的のために、Kafkaは広く使用されているプログラミング言語や環境のための多くのクライアントライブラリを提供しています。

このチュートリアルでは、Kafkaトピックからデータを取得するJavaプログラムを作成します。前回のチュートリアル「CLIを通じてデータソースにKafkaプロデューサーを設定する方法」からのJavaプロジェクトを再利用します。

あなたは、トピックからメッセージを消費することでKafkaクライアントを活用するクラスを実装します。その後、Kafkaがどのように複数のコンシューマが同じトピックを同時に読み取るのを管理し、彼らの進捗をどのように追跡するかを学びます。また、コンシューマの進捗を手動でクラスターに報告する方法も学びます。

CLIを通じてデータを受信するためのKafkaコンシューマーの設定方法

  1. Kafkaコンシューマーの作成
  2. 優雅なシャットダウンの実装
  3. コンシューマグループによるロードバランシング
  4. オフセットを手動でコミットする

前提条件

このチュートリアルを完了するためには、次のものが必要です:

  • 少なくとも4GBのRAMと2つのCPUを備えたマシン。Ubuntuサーバーの場合は、セットアップ手順に従ってUbuntu初期サーバーセットアップを行ってください。
  • Droplet またはローカルマシンに Java Development Kit (JDK) 8 以上がインストールされていること。Ubuntu に Java をインストールする方法については、「Apt を使用して Ubuntu に Java をインストールする方法」チュートリアルを参照してください。
  • Droplet またはローカルマシンに Apache Kafka がインストールされ、設定されました。セットアップ手順については、Kafka の入門チュートリアルに従ってください。
  • Javaプロジェクトの標準ディレクトリレイアウトに関する理解。詳細については、公式Mavenドキュメントの「標準ディレクトリレイアウトの紹介」トピックを参照してください。
  • CLIチュートリアルに従ってデータをソースするために設定されたKafkaプロデューサーを持つJavaプロジェクト。

ステップ1 - Kafkaコンシューマーの作成

前提条件の一環として、Kafkaにプログラムでアクセスし、java_demoトピックにメッセージを生成するための必要な依存関係を持つJavaプロジェクトを作成しました。このステップでは、そのトピックからメッセージを消費するクラスを作成します。

「dokafka」プロジェクトが保存されているディレクトリに移動してください。プロジェクトの構造に従って、ソースコードは「src/main/java/com/dokafka」の下に保存されています。

クラスは ConsumerDemo.java という名前のファイルに保存します。以下のコマンドを実行して作成し、編集のために開いてください:

nano src/main/java/com/dokafka/ConsumerDemo.java

次の行を追加してください:

package com.dokafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.time.*;

public class ConsumerDemo {
  private static final Logger log = LoggerFactory.getLogger(ConsumerDemo.class);

  public static void main(String[] args) {
    String bootstrapServers = "localhost:9092";
    String topicName = "java_demo";
    String groupId = "group1";

    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList(topicName));

    try {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

        for (ConsumerRecord<String, String> record: records) {
          log.info(String.format("topic = %s, partition = %d, offset = %d, value = %s\n",
            record.topic(),
            record.partition(),
            record.offset(),
            record.value()));
        }
      }
    } catch (Exception e) {
      log.error("An error occurred", e);
    } finally {
      consumer.close();
    }
  }
}

まず、ConsumerDemoクラスを定義し、使用するクラスをインポートします。また、クラスのメンバーとしてLoggerをインスタンス化します。ProducerDemoと同様に、main()メソッドでは、最初にKafkaクラスターのアドレス（bootstrapServers）とメッセージを消費するトピックの名前を宣言します。

すべてのKafkaコンシューマーは、グループIDと呼ばれる一意の文字列で識別されるコンシューマーグループに属しています。これをgroup1として定義し、groupIdに保存します。

次に、Kafka コンシューマーを操作するための設定を表すキーと値のペアを保持する Properties オブジェクトをインスタンス化します。BOOTSTRAP_SERVERS_CONFIG プロパティを Kafka クラスターのアドレスに設定します。ここでは、キーおよび値の デシリアライザ エントリを StringDeserializer.class.getName() に設定します。

シリアライザー（ProducerDemoクラスで使用される）とは逆に、デシリアライザーはバイト入力を受け取り、元のオブジェクトを再構築するクラスです。コンシューマは、キーと値のネットワークで受け入れ可能な状態を、コードが理解する元の形に戻すためにそれらを使用します。キーと値の両方は、組み込みのStringDeserializerを使用して文字列にデシリアライズされます。

次に、消費者のグループID（GROUP_ID_CONFIG）を設定します。また、AUTO_OFFSET_RESET_CONFIGパラメータも設定します。これは、消費者が以前に保存された位置がない場合、トピックのどこから読み始めるべきかを定義します。これをearliestに設定すると、最初から（オフセット0）読み始めるよう指示します。

その後、KafkaConsumerを宣言してインスタンス化します。

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

消費者は、構成のための付随するプロパティとともに、String型のキーと値を解析します。

topicからメッセージを受け取りたい場合は、subscribe()します:

consumer.subscribe(Collections.singletonList(topicName));

消費者は、各トピックのIDをListの形式で渡すことにより、一度に複数のトピックに購読することができます。あなたは一つのトピックにしか購読しないため、Collections.singletonList()ヘルパーメソッドを使用して、topicを唯一の要素とするListを作成します。

次に、無限ループでポーリングしてトピックレコードを受け取り始めます:

...
    try {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

        for (ConsumerRecord<String, String> record: records) {
          log.info(String.format("topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
            record.topic(),
            record.partition(),
            record.offset(),
            record.key(),
            record.value()));
        }
      }
    } catch (Exception e) {
      log.error("An error occurred", e);
    } finally {
      consumer.close();
    }
...

KafkaConsumer の poll() メソッドは Duration を受け入れ、これは新しいレコードがストリーミングされるまでコンシューマが待機する時間を示します。その後、受信した各レコードのメタデータと値をログに記録します。無限ループが try ブロック内にあることに注意してください。これがなければ、consumer.close() に到達できないため、コンパイルが失敗します。

終わったら、ファイルを保存して閉じてください。

次に、ConsumerDemoをコンパイルして実行するスクリプトを作成します。それをrun-consumer.shというファイルに保存します。作成して編集のために開いてください:

nano run-consumer.sh

次の行を追加してください:

#!/bin/bash
mvn clean
mvn package
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.ConsumerDemo

ファイルを保存して閉じてから、それを実行可能としてマークしてください:

chmod +x run-consumer.sh

最後に、それを実行してみてください。

./run-consumer.sh

出力は長くなるので、その終わりはこのように見えるはずです:

...
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Notifying assignor about the new Assignment(partitions=[java_demo-0])
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListenerInvoker - [Consumer clientId=consumer-group1-1, groupId=group1] Adding newly assigned partitions: java_demo-0
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Found no committed offset for partition java_demo-0
[main] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=consumer-group1-1, groupId=group1] Resetting offset for partition java_demo-0 to position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}.
[main] INFO com.dokafka.ConsumerDemo - topic = java_demo, partition = 0, offset = 0, key = null, value = Hello World!
[main] INFO com.dokafka.ConsumerDemo - topic = java_demo, partition = 0, offset = 1, key = null, value = Hello World!

KafkaConsumer は、その java_demo トピックのパーティション 0 に割り当てられたことをログに記録し、そのパーティションの コミットされたオフセット を見つけられなかったことを示しました。そのため、以前に AUTO_OFFSET_RESET_CONFIG を earliest に設定したので、パーティションのオフセットは次のログメッセージで示されているように 0 にリセットされます。コンシューマは、オフセットを定期的（または手動で）クラスターにコミットすることで、読み取りを停止した場所を追跡します。オフセットとそれがパーティションやコンシューマグループにどのように関連しているかについては、次のセクションで詳しく学びます。

最初から読み始めたので、消費者はあなたが前提条件の一部として生成したHello World!メッセージを受け取りました。

停止するには CTRL+C を押し、その後再度実行してください:

...
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerUtils - Setting offset for partition java_demo-0 to the committed offset FetchPosition{offset=2, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}
...

消費者は現在、オフセット2から読み始めます。これは、前回停止した場所です。つまり、Kafkaクライアントは以前のオフセットをクラスターに自動的に保存したことを意味します。

このステップでは、`java_demo` トピックからレコードをストリーミングする Kafka コンシューマを Java で作成しました。次に、シャットダウン時に `KafkaConsumer` を適切に処理するように拡張します。

ステップ2 - 優雅なシャットダウンの実装

最後のブロック内の consumer.close() 行は、実行の流れが try ブロックを終了した後に実行されます。無限ループを含んでいるため、それは例外が発生した場合にのみ起こります。これから、ConsumerDemo を拡張して、消費者がシャットダウンまたは終了するときにも消費者を閉じるようにします。

Javaでは、プログラムが終了する際に実行される関数を登録できる機能があり、これをシャットダウンフックと呼びます。ConsumerDemoを編集するために開いてください:

nano src/main/java/com/dokafka/ConsumerDemo.java

ハイライトされた行を追加してください:

...
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList(topic));

    Thread currentThread = Thread.currentThread();
    Runtime.getRuntime().addShutdownHook(new Thread() {
      public void run() {
        consumer.wakeup();

        try {
          currentThread.join();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    });
...

ここでは、現在実行中のスレッドへの参照を最初に取得します。次に、新しいThreadを渡すことでシャットダウンフックを追加します。そのスレッドのrun()メソッドをオーバーライドし、コンシューマにwakeup()を命令します。これにより、コンシューマが停止し、WakeupExceptionが発生するので、これをキャッチする必要があります。その後、finallyブロックが実行され、コンシューマがclose()されます。それを実現するために、このスレッドはメインのスレッドとマージし、実行の流れをメインのスレッドに戻します。

次の行を try ブロックの後に追加して WakeupException を catch する必要があります:

...
    } catch (WakeupException e) {
      // Ignore
    } catch (Exception e) {
      log.error("An error occurred", e);
    } finally {
      consumer.close();
    }
...

WakeupExceptionは、消費者がfinallyで閉じられるため、現在はキャッチされて無視されています。

作業が完了したらファイルを保存して閉じ、その後コンシューマーを実行します:

./run-consumer.sh

読み込みが完了するまで待ってから、CTRL+Cを押してください。シャットダウンが即座に行われないことに気付くでしょう。これは、コンシューマがクラスターと通信し、コンシューマグループからの離脱を通知しているためです。このことを示す出力の終わりは、以下のようになります:

...
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListenerInvoker - [Consumer clientId=consumer-group1-1, groupId=group1] Revoke previously assigned partitions java_demo_partitions-0, java_demo_partitions-1
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Member consumer-group1-1-7bddcea2-ee1a-4a15-9797-669c0302d19f sending LeaveGroup request to coordinator localhost:9092 (id: 2147483646 rack: null) due to the consumer is being closed
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Resetting generation and member id due to: consumer pro-actively leaving the group
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Request joining group due to: consumer pro-actively leaving the group
...

このステップでは、プログラムがシャットダウンされる際にコンシューマが適切に閉じることを保証するために、ConsumerDemoにシャットダウンフックを実装しました。これから、コンシューマグループとオフセットのアーキテクチャについてさらに学びます。

ステップ3 - 消費者グループによる負荷分散

前のステップで作成した消費者のGROUP_ID_CONFIGパラメータはgroup1に設定されており、これはそのIDを持つ消費者グループに属していることを示します。グループ内の消費者は同質と見なされ、Kafkaはそれらの間でイベントを負荷分散することができます。

トピックパーティション

各Kafkaトピックはパーティションで構成されており、これはクラスター内のノードに分散されて障害耐性を提供します。各パーティションは独自のレコードのセットとオフセットを持っています。Kafkaはパーティション内のレコードの順序を保証しますが、トピック内のパーティション間での順序は保証されません。デフォルトでは、Kafkaは受信したレコードがどのパーティションに追加されるべきかを決定するためにラウンドロビンアルゴリズムを使用します。一般的には有用ですが、関連するレコードを強い順序で保存する必要がある場合には受け入れられません。なぜなら、1つのパーティションを読み取るだけでは、全体の論理的なイベントの連鎖を得ることができないからです。

厳密な順序を維持するために、Kafkaはメッセージのためにキーを渡すことを許可します。キーを使用すると、Kafkaは常にレコードを同じパーティションに追加します。Kafkaはパーティション内の順序を保証するため、同じキーを持つすべてのメッセージが挿入の順序でストリームされることが確実です。

パーティションは冗長性を確保するために複数のブローカーにレプリケートされることができます。消費者が相互作用する主なパーティションは パーティションリーダー と呼ばれ、二次的なものは レプリカ とみなされます。レプリカは、パーティションリーダーが存在するノードとは異なるノードに保存されます。

消費者グループとパーティションのマッピング

各パーティションは独自のオフセットを持つイベントの別々のストリームであるため、各パーティションはグループ内の消費者に割り当てられます。パーティション数が消費者数より多い場合、一部の消費者は複数のパーティションから読み取ることになります。逆に、消費者数がパーティション数より多い場合、一部の消費者には何も割り当てられず、アイドルのままとなります。これは意図的に行われており、複数の消費者が同じパーティションを並行して読み取ると、同じイベントが複数回処理されることにつながるためです。この理由から、グループ内のパーティション数と消費者数を一致させることが推奨されます。

消費者はいつでも消費者グループに参加したり、離れたりできます。その際、Kafkaは新しい消費者のセットの間でパーティションを再バランスします。その後、各消費者は、自分に割り当てられた各パーティションの最新のコミット済みオフセットを取得し、そこから処理を続けます。消費者が以前に作業していたパーティションのセットを保持するかどうかについては保証がないため、作業が実際に完了したときにのみオフセットをKafkaにコミットすることが非常に重要です。

レコードのためのキー設定

あなたは今、2つのパーティションを持つ新しいトピックを作成します。次に、ProducerDemoを修正して、送信する各メッセージにキーを設定し、複数のコンシューマをグループで実行して、Kafkaが受信メッセージをどのように配置するかを確認します。

あなたのKafkaインストールディレクトリから、次のコマンドを実行して新しいトピックを作成してください。

bin/kafka-topics.sh --create --topic java_demo_partitions --bootstrap-server localhost:9092 --partitions 2

このトピックはjava_demo_partitionsと呼ばれ、2つのパーティションが含まれています。出力は次のとおりです:

...
Created topic java_demo_partitions.

あなたはProducerDemoを修正するので、編集のためにそれを開いてください:

nano src/main/java/com/dokafka/ProducerDemo.java

最初に、topicNameをjava_demo_producerに設定します:

...
    String topicName = "java_demo_partitions";
...

それから、プログラムが受け取る最初のCLI引数をProducerRecordのキーとして渡します:

...
    ProducerRecord<String, String> producerRecord =
      new ProducerRecord<>(topicName, args[0], "Hello World!");    
...

文字列配列 args は main() メソッドに渡され、プログラムを実行する際に渡される引数が含まれています。作業が完了したら、ファイルを保存して閉じてください。

次に、ConsumerDemoを更新して新しいトピックからデータを取得します。編集のために開いてください:

nano src/main/java/com/dokafka/ConsumerDemo.java

同様に、topicNameを新しいトピックに設定します。

...
    String topicName = "java_demo_partitions";
...

ファイルを保存して閉じてください。

プロデューサーを実行する前に、与えられた引数をProducerDemoに渡すために、run-producer.shスクリプトを更新する必要があります。編集のために開いてください:

nano run-producer.sh

以下のようにjavaコマンドに最初の引数を渡してください:

#!/bin/bash
mvn clean
mvn package
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.ProducerDemo $1

終わったら、ファイルを保存して閉じてください。

次に、別のターミナルセッションを開き、コンシューマーを実行します:

./run-consumer.sh

次に、3つ目のターミナルセッションを開き、そこに2番目のコンシューマを実行します:

./run-consumer.sh

消費者の一人の出力の最後は次のようになります:

...
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Notifying assignor about the new Assignment(partitions=[java_demo_partitions-0])
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListenerInvoker - [Consumer clientId=consumer-group1-1, groupId=group1] Adding newly assigned partitions: java_demo_partitions-0
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerUtils - Setting offset for partition java_demo_partitions-0 to the committed offset FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}

別の消費者にはjava_demo_partitions-1パーティションが割り当てられます。

メインセッションでは、次のように実行して key1 のメッセージを生成します:

./run-producer.sh key1

二人の消費者のうち、一人だけがメッセージを受け取ることに注意してください:

[main] INFO com.dokafka.ConsumerDemo - topic = java_demo_partitions, partition = 0, offset = 0, key = key1, value = Hello World!

異なるキーでメッセージを生成してみてください:

./run-producer.sh key2

今回は、別の消費者がこのメッセージを受け取ります。なぜなら、Kafkaがその消費者がレコードをストリーミングしている残りのパーティションにルーティングしたからです。

[main] INFO com.dokafka.ConsumerDemo - topic = java_demo_partitions, partition = 1, offset = 0, key = key2, value = Hello World!

3つ目のセッションでCTRL+Cを押して2つ目のコンシューマを終了させてください。残りのコンシューマに対して2つのパーティションが再バランスされるのが表示されるでしょう:

...
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-group1-1, groupId=group1] Notifying assignor about the new Assignment(partitions=[java_demo_partitions-0, java_demo_partitions-1])
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListenerInvoker - [Consumer clientId=consumer-group1-1, groupId=group1] Adding newly assigned partitions: java_demo_partitions-0, java_demo_partitions-1
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerUtils - Setting offset for partition java_demo_partitions-1 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerUtils - Setting offset for partition java_demo_partitions-0 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 1 rack: null)], epoch=0}}

では、再度メッセージを生成してみてください:

./run-producer.sh key1
./run-producer.sh key2

残りの消費者が両方を受け取ったことがわかります:

...
[main] INFO com.dokafka.ConsumerDemo - topic = java_demo_partitions, partition = 0, offset = 1, key = key1, value = Hello World!

[main] INFO com.dokafka.ConsumerDemo - topic = java_demo_partitions, partition = 1, offset = 1, key = key2, value = Hello World!

このステップでは、グループ内で複数のコンシューマーを実行することと、変更があった場合にKafkaがパーティションを再割り当てすることでトラフィックのバランスを再調整する方法について学びました。次に、オフセットを手動でクラスターにコミットする方法について学びます。

ステップ4 - オフセットを手動でコミットする

Kafkaクライアントライブラリのデフォルトの動作は、poll()によって返された最新のオフセットを5秒ごとに自動的にコミットすることです。これは、消費者がイベントを自動コミットよりも遅いペースで処理する場合には安全ではなく、レコードが未処理のまま残る可能性があります。これから、問題を防ぐためにオフセットを手動でコミットする方法を学びます。

たとえば、poll() がオフセット 0 から 10 までのメッセージを返した場合、コンシューマは、実際にその処理が行われたかどうかにかかわらず、10 が最新の処理済みオフセットであると返します。アプリがオフセット 10 のメッセージを実際に処理する前にクラッシュした場合、次回の実行ではそこから開始するため、一部のレコードが未処理のまま残ります。

あなたはConsumerDemoを修正するので、編集のためにそれを開いてください:

nano src/main/java/com/dokafka/ConsumerDemo.java

まず、ENABLE_AUTO_COMMIT_CONFIGをfalseに設定して自動コミットを無効にします:

...
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
...

次に、受信した各レコードをループする部分にハイライトされた行を追加します:

...
    try {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

        for (ConsumerRecord<String, String> record: records) {
          log.info(String.format("topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
            record.topic(),
            record.partition(),
            record.offset(),
            record.key(),
            record.value()));
          consumer.commitSync();
        }
      }
    }
...

KafkaConsumerのcommitSync()メソッドは、現在のオフセットをブロッキング方式でコミットします。つまり、コミットプロセスが終了するか例外がスローされるまで、レコードのさらなる処理は行われません。

KafkaConsumerは、commitAsync()メソッドも提供しており、これは実行の流れを呼び出し元に戻しながら（この場合、ループは引き続き実行されます）バックグラウンドでコミットしようとします。この方法の欠点は、状態が依然として不整合である可能性があることです。Kafkaがエラーを返すと、あなたのコードは次のレコードに進んでしまうかもしれません。

作業が完了したら、ファイルを保存して閉じてください。別のセッションでコンシューマを実行して、いくつかのメッセージを生成してみることができます。全体の流れは同じですが、コンシューマはまだ処理していないオフセットをコミットしません。

このセクションでは、受信したレコードを処理する際にオフセットを手動でKafkaにコミットする方法を学びました。

結論

このチュートリアルでは、ProducerDemo Kafkaプロデューサーを拡張し、Javaで書かれたKafkaコンシューマであるConsumerDemoを作成しました。消費者グループについて、Kafkaがどのようにパーティションをグループ内の消費者に割り当てるかについて学びました。

あなたは、プロセスが終了しているときでさえ、KafkaConsumerインスタンスが適切に閉じられることを確認しました。また、レコードを処理しながらオフセットを手動でクラスターにコミットする方法も学びました。

KafkaConsumerおよびそのプロパティに関する詳細情報については、公式のKafkaドキュメントを訪問してください。

著者はWrite for DOnationsプログラムの一環として、寄付を受け取るためにオープンソースイニシアティブを選択しました。

関連記事: