カフカコンシューマのメッセージを回復する方法

Question

私はApache camelでアプリケーションを書いています。私はいくつかのカフカのトピックからキャメルカフカのコンポーネントを介してメッセージを消費しており、クラッシュ/再起動が発生した場合の回復のためにデータベースにダンプします。以下はラクダであるURI

カフカ：？autoCommitEnable = falseを&のgroupId = R & keySerializerClass = org.apache.kafka.common.serialization.StringSerializer & serializerClass = org.apache.kafka.common.serialization.StringSerializer &トピック=

私のユースケースがある - 私はカフカからの何らかのメッセージ（複数可）を消費しているが、回復のためにデータベースに同じことをダンプと同じで、すべての失われたメッセージを取得する方法をhappens.Nowをクラッシュことができませんでしたアプリケーションを再起動した後のコンシューマグループID？ ありがとう

Answer 1

アプリケーションを再起動した後、同じコンシューマグループIDのすべての失われたメッセージを取得する方法はありますか？

実際にkafkaは、あなたのアプリケーションでオフセットをコミットする場合、コンシューマオフセットを格納します。したがって、アプリケーションを再起動すると、カフカに保存されている最後のオフセットからのメッセージが消費されます。

あなたは、私はまた、このhttps://github.com/apache/camel/blob/camel-2.18.2/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.javaを見つけた真のAutocommitEnable = OR

を設定することができます。

いくつかの作品のコードがあります

if (endpoint.getConfiguration().isAutoCommitEnable() != null 
         && !endpoint.getConfiguration().isAutoCommitEnable()) { 
         long partitionLastoffset = partitionRecords.get(partitionRecords.size() - 1).offset(); 
         consumer.commitSync(Collections.singletonMap(
          partition, new OffsetAndMetadata(partitionLastoffset + 1))); 
        } 

キャメルも、あなたはAutocommitEnableを設定していないこのの世話をします。