Aiven-Open / Aiven-Open/opensearch-connector-for-apache-kafka

batches limited to 500 messages

オープン
#168 コメント 2 件 リアクション 0 件 担当者 0 名 GitHub で見る
主要言語
Java
スター
95
フォーク
61
平均マージ
11時間 52分
マージ済み PR(30日)
7

説明

I'm working with version 2.0.4 of the opensearch connector, to sink to AWS Opensearch cluster version 2.3.
We've activated TRACE level for org.apache.kafka.connect.runtime.WorkerSinkTask, and we're seeing that the batch size is never above 500, even when the specific consumer for that partition is tens of thousands of messages behind.
This is my elasticsearch sink configuration:
{
"connector.class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
"type.name": "_doc",
"behavior.on.null.values": "delete",
"tasks.max": "24",
"connection.timeout.ms": "3000",
"max.retries": "10",
"key.ignore": "false",
"retry.backoff.ms": "1000",
"max.buffered.records": "100000",
"errors.deadletterqueue.topic.replication.factor": "1",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter": "org.apache.kafka.connect.converters.IntegerConverter",
"read.timeout.ms": "10000",
"behavior.on.version.conflict": "warn",
"topics": "SEARCH",
"batch.size": "10000",
"max.in.flight.requests": "25",
"schema.ignore": "false",
"behavior.on.malformed.documents": "fail",
"value.converter.schema.registry.url": "http://cp-schema-registry:8081",
"flush.timeout.ms": "20000",
"errors.deadletterqueue.topic.name": "dlq_search",
"name": "ELASTICSEARCH_SINK",
"errors.tolerance": "all",
"connection.url": "https://....",
"linger.ms": "1000"
}

Is there some hard limitation on the batch size of 500? Even when I change the config value of batch.size to 400, I see in the trace "batch.size = 400 ....Delivering batch of 500 messages to task". Can you help me understand how to control the batch size?

Thank you.

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。