# Kafka connect to ES **URL:** \ **Category:** Kafka Connect\ **Created:** [4 February 2021 17:56 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645 "2021-02-04T17:56:03Z")\ **Posts on this page:** 10\ **Page:** 1
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [4 February 2021 17:56 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/1 "2021-02-04T17:56:03Z")
Hello I am pretty new to kafka connect only about 2 weeks in of using it so bare with me. Our environment is all in the AWS. So we have Kafka brokers and ES in AWS. We are currently trying to send data from sql server to ES. We have gotten all the data into a producer and now trying to stream the data to ES. I am currently running in distributed mode with a ES sink connector. Once I run that seems to start pushing data into ES pretty quickly but then kinda get stuck. I am able to query the data that does come in though. This is where it usually get stuck. > INFO WorkerSinkTask{id=elastic-search-smb-2-0} Committing offsets asynchronously using sequence number 5: {color.smd.abs-9=OffsetAndMetadata{offset=780831, leaderEpoch=null, metadata=‘’}, color.smd.abs-7=OffsetAndMetadata{offset=787143, leaderEpoch=null, metadata=‘’}, color.smd.abs-8=OffsetAndMetadata{offset=781276, leaderEpoch=null, metadata=‘’}, color.smd.abs-5=OffsetAndMetadata{offset=780728, leaderEpoch=null, metadata=‘’}, color.smd.abs-6=OffsetAndMetadata{offset=779078, leaderEpoch=null, metadata=‘’}, color.smd.abs-3=OffsetAndMetadata{offset=781514, leaderEpoch=null, metadata=‘’}, color.smd.abs-4=OffsetAndMetadata{offset=782010, leaderEpoch=null, metadata=‘’}, color.smd.abs-1=OffsetAndMetadata{offset=781538, leaderEpoch=null, metadata=‘’}, color.smd.abs-2=OffsetAndMetadata{offset=778608, leaderEpoch=null, metadata=‘’}, color.smd.abs-0=OffsetAndMetadata{offset=777293, leaderEpoch=null, metadata=‘’}} (org.apache.kafka.connect.runtime.WorkerSinkTask:346) After a little time it will then throw this error below: > sending LeaveGroup request to coordinator b-2.\*\*\*\*\*.amazonaws.com:9092 due to consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records. Not sure if its to much data being pushed through or something wrong with my configuration. Also not sure what config file where this max.poll.interval can be added to. Any help would be much appreciated. ---
**Author:** ![mcascallares](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mcascallares/32/274_2.png) [@mcascallares](https://forum.confluent.io/u/mcascallares)\ **Post date:** [4 February 2021 18:34 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/2 "2021-02-04T18:34:43Z")
Hi, In a nutshell, what is happening is that time required to send each batch to ES is longer than the time between subsequent Kafka poll calls and the heartbeat does not arrive on time. As it’s mentioned in the log message, you have to either increase `max.poll.interval.ms` or make smaller batches when sending data to ES. For this, you can specify any of them in your connector configuration. The property keys are, respectively: `consumer.max.poll.interval.ms` `batch.size` More information here: [https://docs.confluent.io/kafka-connect-elasticsearch/current/configuration\_options.html#](https://docs.confluent.io/kafka-connect-elasticsearch/current/configuration_options.html#) Hope this helps! ---
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [4 February 2021 18:41 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/3 "2021-02-04T18:41:40Z")
Thank you. So this would go in the ES connector when i create that correct? Also i don’t see `consumer.max.poll.interval.ms` in the doc. Would it be `connection.timeout.ms`? ---
**Author:** ![mcascallares](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mcascallares/32/274_2.png) [@mcascallares](https://forum.confluent.io/u/mcascallares)\ **Post date:** [4 February 2021 18:51 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/4 "2021-02-04T18:51:08Z")
That setting is not in the ES Sink Connector, but in the common consumer Connector configuration (you can imagine it like a parent class). More information here: [https://docs.confluent.io/home/connect/userguide.html#producer-and-consumer-overrides](https://docs.confluent.io/home/connect/userguide.html#producer-and-consumer-overrides) In any case, the key is the one I mentioned `consumer.max.poll.interval.ms` ---
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [4 February 2021 19:02 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/5 "2021-02-04T19:02:00Z")
Sorry im still kind of confused. When I run the worker I use bin/connect-distributed etc/kafka/connect-distributed.properties Does it mean i need to add `consumer.max.poll.interval.ms` to the connect-distributed.properties file? ---
**Author:** ![mcascallares](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mcascallares/32/274_2.png) [@mcascallares](https://forum.confluent.io/u/mcascallares)\ **Post date:** [4 February 2021 19:02 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/6 "2021-02-04T19:02:38Z")
Yep, that’s correct 🙂 ---
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [4 February 2021 19:08 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/7 "2021-02-04T19:08:31Z")
Sorry two more questions. Anything to do with the consumer I need to put that in front. so for \> batch.size I need to use \> consumer.batch.size. Is that correct? Also, I am running this bin/connect-distributed -daemon. Is this the best way to run it in the background or is like a service better? How do users usually do this in production? If you do run it as a daemon whats the best way to stop the worker? Thank you again you have been a really great help. ---
**Author:** ![mcascallares](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mcascallares/32/274_2.png) [@mcascallares](https://forum.confluent.io/u/mcascallares)\ **Post date:** [5 February 2021 10:17 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/8 "2021-02-05T10:17:46Z")
`batch.size` is not a consumer setting but a specific setting define by the ES Sink Connector, no `consumer` prefix is needed. More details [here](https://docs.confluent.io/kafka-connect-elasticsearch/current/configuration_options.html). For production environments, it’s recommended to run Kafka Connect in distributed mode. You can do it in the background or as a service, up to you. To stop the process, send a kill signal to the connect process `kill `. ---
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [5 February 2021 14:44 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/9 "2021-02-05T14:44:18Z")
Great thank you. So the `batch.size` is part of the ES config connector when you create that. The `consumer.max.poll.interval.ms` goes in the connect-distributed.properties. Again thank you for the quick responses you been very helpful as I am new to kafka. ---
**Author:** ![robbyd](https://avatars.discourse-cdn.com/v4/letter/r/5e9695/32.png) [@robbyd](https://forum.confluent.io/u/robbyd)\ **Post date:** [8 February 2021 14:42 UTC](https://forum.confluent.io/t/kafka-connect-to-es/645/10 "2021-02-08T14:42:09Z")
So I was able to add those new parameters and I do not get the error anymore. It seems like it works for the first 30 seconds and sends lots of data but then after that it seems to get stuck. I don’t see any error though. Only thing I see in the log is this. `INFO WorkerSinkTask{id=elastic-search-smb-4-0} Committing offsets asynchronously using sequence number 6: {color.smd.abs-9=OffsetAndMetadata{offset=764214, leaderEpoch=null, metadata=''}, color.smd.abs-7=OffsetAndMetadata{offset=772233, leaderEpoch=null, metadata=''}, color.smd.abs-8=OffsetAndMetadata{offset=764626, leaderEpoch=null, metadata=''}, color.smd.abs-5=OffsetAndMetadata{offset=764059, leaderEpoch=null, metadata=''}, color.smd.abs-6=OffsetAndMetadata{offset=764335, leaderEpoch=null, metadata=''}, color.smd.abs-3=OffsetAndMetadata{offset=765716, leaderEpoch=null, metadata=''}, color.smd.abs-4=OffsetAndMetadata{offset=765397, leaderEpoch=null, metadata=''}, color.smd.abs-1=OffsetAndMetadata{offset=765028, leaderEpoch=null, metadata=''}, color.smd.abs-2=OffsetAndMetadata{offset=763931, leaderEpoch=null, metadata=''}, color.smd.abs-0=OffsetAndMetadata{offset=763980, leaderEpoch=null, metadata=''}} (org.apache.kafka.connect.runtime.WorkerSinkTask:346)` Any idea what else can be done to try and get this data into ES. I set the batch to 100 but still doesn’t seem to work. Could it be something in my ES config file? Does `tasks.max` effect anything in the connector?