181.7. Kafka コンシューマーでの手動コミットの使用

Camel 2.21 以降で利用可能

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

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

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.commitSync();
}

これにより、同期コミットが強制され、Kafka でコミットが確認されるまでブロックされるか、失敗した場合は例外が出力されます。

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