コンフルエントプラットフォームのKafka-Connect APIでmax.poll.recordsを設定する方法

Question

コンフルエント3.0.1プラットフォームを使用し、Kafka-Elasticsearchコネクタを構築しています。このために私はKafkaからデータを取得するためにSinkConnectorとSinkTask（Kafka-connect APIs）を拡張しています。

このコードの一部として、一度に100個のレコードしか取得できないように、「max.poll.records」を返すようにSinkConnectorのtaskConfigsメソッドを拡張しています。しかし、それは動作していないと私はすべてのレコードを同時に取得しており、私は規定された時間内にオフセットをコミットしていません。いずれかは、私が「max.poll.records」

public List<Map<String, String>> taskConfigs(int maxTasks) { 
    ArrayList<Map<String, String>> configs = new ArrayList<Map<String, String>>(); 
    for (int i = 0; i < maxTasks; i++) { 
     Map<String, String> config = new HashMap<String, String>(); 
     config.put(ConfigurationConstants.CLUSTER_NAME, clusterName); 
     config.put(ConfigurationConstants.HOSTS, hosts); 
     config.put(ConfigurationConstants.BULK_SIZE, bulkSize); 
     config.put(ConfigurationConstants.IDS, elasticSearchIds); 
     config.put(ConfigurationConstants.TOPICS_SATELLITE_DATA, topics); 
     config.put(ConfigurationConstants.PUBLISH_TOPIC, topicTopublish); 
     config.put(ConfigurationConstants.TYPES, elasticSearchTypes); 
     config.put("max.poll.records", "100"); 

     configs.add(config); 
    } 
    return configs; 
    } 

Answer 1

あなたは、コネクタの構成でmax.poll.recordsようなほとんどのカフカの消費者のconfigsをオーバーライドすることはできませんを設定することができますしてください。 Connectワーカーの設定では、consumer.という接頭辞を使用することができます。

Answer 2

これを解決しました。私は以下の設定をconnect-avro-standalone.propertiesに追加しました。

group.id=mygroup 
consumer.max.poll.records=1000 

そして、以下のコマンドを実行してマイナーコネクタを実行しました。

sh ./bin/connect-standalone ./etc/schema-registry/connect-avro-standalone.properties ./etc/kafka-connect-elasticsearch/connect-elasticsearch-sink.properties