34.10. Kafka consumer での手動コミットの使用

デフォルトでは、Kafka consumer は自動コミットを使用します。オフセットは、指定された間隔を使用してバックグラウンドで自動的にコミットされます。

手動コミットを強制する場合は、メッセージヘッダーに保存されている Camel Exchange の KafkaManualCommit API を使用できます。これには、KafkaComponent またはエンドポイントでオプション allowManualCommit を true に設定して、手動コミットを有効にする必要があります。次に例を示します。

KafkaComponent kafka = new KafkaComponent();
kafka.setAllowManualCommit(true);
...
camelContext.addComponent("kafka", kafka);

KafkaComponent kafka = new KafkaComponent();
kafka.setAllowManualCommit(true);
...
camelContext.addComponent("kafka", kafka);

その後、Camel Processor などの Java コードから KafkaManualCommit を使用できます。

public void process(Exchange exchange) {
    KafkaManualCommit manual =
        exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
    manual.commit();
}

public void process(Exchange exchange) {
    KafkaManualCommit manual =
        exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
    manual.commit();
}

これにより、同期コミットが強制され、Kafka でコミットが確認されるまでブロックされるか、失敗した場合は例外が出力されます。KafkaManualCommitFactory を `DefaultKafkaManualAsyncCommitFactory` 実装で設定することにより、非同期コミットも使用できます。

その後、kafka 非同期コミット API を使用して、次の consumer ループでコミットが行われます。パーティションからのレコードは、一意のスレッドによって処理およびコミットする必要があることに注意してください。そうでない場合、一貫性のない動作が発生する可能性があります。これは、集約の完了タイムアウト戦略で最も役立ちます。

KafkaManualCommit のカスタム実装を使用する場合は、カスタム実装のインスタンスを作成する KafkaComponent でカスタム KafkaManualCommitFactory を設定できます。