6.3. コンシューマーとの協力的なリバランスの使用
Kafka コンシューマーは、適切なリバランスプロトコルによって決定されるパーティション割り当て戦略を使用します。デフォルトでは、Kafka は RangeAssignor プロトコルを採用します。これにより、コンシューマーはリバランス中にパーティション割り当てを放棄することになり、サービスが中断される可能性があります。
効率を向上させ、ダウンタイムを短縮するには、協調的なリバランスアプローチである CooperativeStickyAssignor プロトコルに切り替えることができます。デフォルトのプロトコルとは異なり、協調リバランスでは、コンシューマーが連携して作業することができ、リバランス中にパーティションの割り当てを保持し、コンシューマーグループ内のバランスを達成するために必要な場合にのみパーティションを解放します。
手順
コンシューマー設定では、
partition.assignment.strategyプロパティーを使用して、プロトコルとしてCooperativeStickyAssignorを使用するように切り替えます。たとえば、現在の設定がpartition.assignment.strategy=RangeAssignor, CooperativeStickyAssignorの場合は、partition.assignment.strategy=CooperativeStickyAssignorに更新します。コンシューマー設定ファイルを直接変更する代わりに、コンシューマーアプリケーションコードで
props.putを使用してパーティション割り当て戦略を設定することもできます。# ... props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"); # ...
- グループ内の各コンシューマーを一度に 1 つずつ再起動し、再起動するたびにグループに再参加できるようにします。
CooperativeStickyAssignor プロトコルに切り替えた後、コンシューマーのリバランス中に RebalanceInProgressException が発生し、同じコンシューマーグループ内の複数の Kafka クライアントが予期せず停止する可能性があります。さらに、この問題により、Kafka コンシューマーがリバランス中にパーティションの割り当てを変更していない場合でも、コミットされていないメッセージが重複する可能性があります。自動オフセットコミット (enable.auto.commit=true) を使用している場合は、何も変更する必要はありません。オフセットを手動でコミットしており (enable.auto.commit=false)、手動コミット中に RebalanceInProgressException が発生した場合は、コンシューマーの実装を変更して、次のループで poll() を呼び出してコンシューマーのリバランスプロセスを完了します。詳細は、カスタマーポータルの CooperativeStickyAssignor の記事を参照してください。
