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

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

コンシューマーには、その動作の側面を制御する多くの構成設定があります。 以下の設定は、最も重要な設定の一部です。

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

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

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

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

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

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

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

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

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

コンシューマーの活動性

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

Consumer.poll(...) から返されるメッセージのバッチが大きい場合、または処理に時間がかかる場合、 poll() が再び呼び出されるまでの遅延は、重大なものや予測不能なものになる可能性があります。 場合によっては、長い最大ポーリング間隔を構成して、メッセージ処理に時間がかかっているというだけの理由でコンシューマーがグループから削除されないようにする必要があります。 このメカニズムが使用可能な唯一のメカニズムである場合、失敗したコンシューマーの検出にかかる時間も長くなります。

コンシューマーの活動性をより簡単に処理できるようにするため、バックグラウンドでのハートビート処理が 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」コールバックが通知された場合は、 ConsumerRebalanceリスナーを使用してオフセットを手動でコミットし (自動コミットを使用していない場合)、「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 で確認してください。

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