メッセージのコンシューム

コンシューマー・プロパティーの構成

コンシューマには、その動作の側面を制御する多くのコンフィギュレーション設定が存在する。 以下の設定は、最も重要な設定の一部です。

さらに多くの構成設定を使用できますが、それらを試す前に、 Apache Kafka の資料 をお読みください。

コンシューマー・グループ

_コンシューマー・グループ_は、1 つ以上のトピックからメッセージをコンシュームするために連携するコンシューマーのグループです。 同じグループ内のすべてのコンシューマーは group.id 構成に同じ値を使用します。 作業負荷を処理するために複数のコンシューマーが必要な場合、同じコンシューマー・グループ内の複数のコンシューマーを実行できます。 1 つのコンシューマーのみが必要な場合でも、通常はgroup.idの値も指定します。

各コンシューマー・グループには、グループ内のコンシューマーへのパーティションの割り当てを担当する_コーディネーター_と呼ばれるサーバーがクラスター内にあります。 コーディネーターの責任は、負荷を平均化するためにクラスター内のすべてのサーバーで負います。 コンシューマーへのパーティションの割り当ては、グループのバランスが再調整されるたびに変化する可能性があります。

コンシューマー・グループにコンシューマーが加わると、そのコンシューマーはグループのコーディネーターを検出します。 その後、コンシューマーはコーディネーターにグループに参加することを通知し、コーディネーターはグループ全体のパーティションのリバランスを開始して新規メンバーを組み込みます。

単一パーティションからのメッセージは、各グループ内の 1 つのコンシューマーによってのみ処理されます。 このメカニズムにより、各パーティション上のメッセージが順番に処理されるようになります。 トピックに 3 つのパーティションが含まれており、そのトピックをコンシュームしているコンシューマー・グループに 2 つのコンシューマーが含まれている例については、以下の図を参照してください。 グループ内の 1 つのコンシューマーには 2 つのパーティションが割り当てられ、もう 1 つのコンシューマーには 1 つのパーティションが割り当てられます。

コンシューマー・グループで以下のいずれかの変更が行われると、変更に対応するためにグループ・メンバーにパーティションの割り当てをシフトすることにより、グループのバランスが再調整されます。

• コンシューマーがグループに参加します。
• コンシューマーがグループを離脱します。
• コンシューマーは、コーディネーターによってもはやライブではないと見なされます。
• 新しいパーティションが既存のトピックに追加されます。

リバランスされたコンシューマー・グループがある場合、グループから離脱したコンシューマーは、グループに復帰するまでコミットが拒否されることに注意してください。 この場合、コンシューマーはグループに再加入する必要があります。その際、以前にコンシューム元にしていたパーティションとは異なるパーティションが割り当てられる可能性があります。

コンシューマーの活動性

Kafka は、障害が起こったコンシューマーを自動的に検出して、動作しているコンシューマーにパーティションを割り当て直します。 ポーリングとハートビートの 2 つのメカニズムを使用します。

If the batch of messages that are returned from Consumer.poll(...) is large or the processing is time-consuming, the delay before poll() is called again can be significant or unpredictable. 場合によっては、長い最大ポーリング間隔を構成して、メッセージ処理に時間がかかっているというだけの理由でコンシューマーがグループから削除されないようにする必要があります。 このメカニズムが使用可能な唯一のメカニズムである場合、失敗したコンシューマーの検出にかかる時間も長くなります。

コンシューマーの活動性をより簡単に処理できるようにするため、バックグラウンドでのハートビート処理が Kafka 0.10.1 で追加されました。 グループのコーディネーターは、グループ・メンバーがアクティブ状態のままであることを示すために定期的にハートビートを送信してくると予期しています。 バックグラウンド・ハートビート・スレッドはコンシューマーで実行され、通常のハートビートをコーディネーターに送信します。 コーディネーターが_セッション・タイムアウト_内のグループ・メンバーからハートビートを受信しない場合、コーディネーターはグループからメンバーを除去し、グループのリバランスを開始します。 セッション・タイムアウトは最大ポーリング間隔よりずっと短くすることができ るので、メッセージ処理に長い時間がかかっても、失敗したコンシューマを検出 するのにかかる時間を短くすることができる。

最大ポーリング間隔を構成するには、max.poll.interval.msプロパティーを使用し、session.timeout.msプロパティーを使用してセッション・タイムアウトを構成します。 メッセージのバッチの処理に 5 分以上かかる場合を除き、これらの設定を使用する必要はありません。

オフセットの管理

コンシューマー・グループごとに、Kafka は、コンシュームされるパーティションごとにコミットされたオフセットを維持します。 コンシューマーは、メッセージを処理してもパーティションからそのメッセージを削除しません。 代わりに、オフセットのコミットと呼ばれるプロセスを使用して、現在のオフセットを更新するだけです。

Event Streams は、コミット済みオフセットの情報を 7 日間保持します。

メッセージをコンシュームする速度の制御

メッセージの洪水に起因するメッセージ処理に問題がある場合、コンシューマー・オプションを設定することで、メッセージの消費速度を制御することができます。 fetch.max.bytes および max.poll.records を使用して、1 回の poll() 呼び出しが返すことのできるデータ量を制御します。

コンシューマーのバランス再調整の処理

コンシューマーがグループに追加されたり、グループから削除されたりすると、グループのリバランスが行われ、コンシューマーはメッセージを消費できなくなる。 その結果、ある消費者グループのすべての消費者が短期間利用できなくなる。

on partitions revoked "コールバックで通知された場合、ConsumerRebalanceListenerを使用してオフセットを手動でコミットし（自動コミットを使用していない場合）、"on partition assigned "コールバックを使用してリバランス成功の通知があるまで以降の処理を一時停止します。

コード・スニペット

これらのコード・スニペットは、関連する概念を説明するための概要レベルです。 完全な例については、GitHub の Event Streams サンプルを参照してください。

コンシューマーを Event Streamsに接続するには、サービス資格情報を作成する必要があります。 これらの資格情報の取得方法については、 Event Streams を参照してください。

コンシューマー・コードでは、まず構成プロパティーのセットをビルドする必要があります。 Event Streamsへの接続はすべて、TLSとユーザーパスワード認証を使ってセキュアに行われるので、少なくともこれらのプロパティが必要です。 BOOTSTRAP_ENDPOINTS、USER、および PASSWORD を、自分のサービス資格情報のものに置き換えます：

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

メッセージを消費するには、以下の例のように、キーと値にデシリアライザを指定する必要もある。

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

これらのデシリアライザーは、プロデューサーによって使用されるシリアライザーと一致する必要があります。

次に、KafkaConsumer を使用してメッセージをコンシュームします。ここでは、各メッセージが 1 つの ConsumerRecord で表されます。 メッセージをコンシュームする最も一般的な方法は、グループ ID を設定することによってコンシューマーをコンシューマー・グループに入れ、次に、トピックのリストに対して subscribe() を呼び出すことです。 コンシューマーにはコンシュームするいくつかのパーティションが割り当てられますが、トピック内のパーティションよりも多くのコンシューマーがグループ内に存在する場合は、コンシューマーにパーティションが割り当てられない可能性があります。 次に、ループ内で poll() を呼び出して、処理するメッセージのバッチを受け取ります。それらのメッセージはそれぞれ 1 つの ConsumerRecord で表されます。

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
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.wakeup() を呼び出すことで他のスレッドから中断させ、整然としたシャットダウンを実現することができる。

オフセットを手動でコミットするには、最初に enable.auto.commit 構成を false に設定する必要があります。 次に、Consumer.commmitSync() または Consumer.commitAsync() のいずれかを使用して、コンシューマーのコミット済みオフセットを定期的に更新します。 単純にするため、この例では、各パーティションのレコードを処理し、最後のオフセットを別個にコミットしています。

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

例外処理

Kafka クライアントを使用する堅固なアプリケーションは、特定の予期される状態の例外を処理する必要があります。 一部のメソッドは非同期であり、 Futureまたはコールバックを使用して結果を送信するため、例外が直接スローされない場合があります。 完全な例を示すコード例を GitHub で確認してください。

コード内の以下の例外リストを処理します。