SimpleConsumer kafkaの自動コミットを無効にする方法0.8.1

Question

kafka SimpleConsumerの自動コミットを無効にしたいと思います。私は0.8.1バージョンを使用しています。上位レベルのコンシューマの場合、configオプションは以下のように設定し、consumerConfigで渡すことができます。 kafka.consumer.Consumer.createJavaConsumerConnector（this.consumerConfig）;

SimpleConsumerで同じようにするにはどうすればよいですか？主に自動コミットを無効にしたい私はconsumer.propertiesでfalseに自動コミットを設定しようとし、kafkaサーバー、zookeeperとproducerを再起動しました。しかし、それは動作しません。私はこの設定をconsumer.propertiesではなく、コードで適用する必要があると思います。 誰でも助けることができますか？ここで

オフセットコミットする前に、コードがクラッシュした上記の場合は私のコードは

List<TopicAndPartition> topicAndPartitionList = new ArrayList<>();
    
topicAndPartitionList.add(topicAndPartition);
 
OffsetFetchResponse offsetFetchResponse = consumer.fetchOffsets(new  OffsetFetchRequest("testGroup", topicAndPartitionList, (short) 0, correlationId, clientName)); 


Map<TopicAndPartition, OffsetMetadataAndError> offsets =  offsetFetchResponse.offsets();


 
FetchRequest req = new FetchRequestBuilder()
.clientId(clientName)
    .addFetch(a_topic, a_partition, offsets.get(topicAndPartition).offset(), 100000) .build(); 

long readOffset = offsets.get(topicAndPartition).offset();

 
FetchResponse fetchResponse = consumer.fetch(req); 

//Consume messages from fetchResponse 


Map<TopicAndPartition, OffsetMetadataAndError > requestInfo = new HashMap<> ();
 
requestInfo.put(topicAndPartition, new OffsetMetadataAndError(readOffset, "metadata", (short)0)); 
OffsetCommitResponse offsetCommitResponse = consumer.commitOffsets(new   OffsetCommitRequest("testGroup", requestInfo, (short)0, correlationId, clientName));
 

どのように見えるかですが、私はまだ最新の私を作る次回の実行中offsets.get（topicAndPartition）.offset（）の結果として、オフセットを取得しますコードが実行されるときにオフセットの自動コミットが発生すると考えること。

Answer 1

SimpleConsumerを使用するということは、オフセットコミットを含むメッセージの消費についてすべてを気にしたいことを意味するだけで、低レベルAPIでは自動コミットがサポートされません。