opened
VincEmond
username_0
I am having some trouble with the Kafka connect here, having some read timeouts with Elastic. It hits the 30 seconds timeout, which is set in the configuration of the connector. When going directly into the docker container after the connector dies and trying to ping Elasticsearch, this HEAD call returns 200 OK without issue and really fast. This issue happens a while after I started the container, basically I start to ingest some data, then the connector stops, and then it will try to restart consuming later. I believe Elastic never receives the call, since it is able to response to this kind of HTTP request really quickly.

Are you aware of any config that I could set so that the connection is always reused with Elastic?
Otherwise, is it possible to make this task recoverable and let it retry later?

```
{"@timestamp":"2020-01-31T15:02:49.645Z","log.level":"ERROR","message":"WorkerSinkTask{id=consumer_group-1} Task threw an uncaught and unrecoverable exception","log.logger":"org.apache.kafka.connect.runtime.WorkerTask","log.type":"kafka-connect-application","service.name":"my-super-java-app"}
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:560)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.connect.errors.ConnectException: java.net.SocketTimeoutException: Read timed out
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.indexExists(JestElasticsearchClient.java:284)
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.createIndices(JestElasticsearchClient.java:290)
	at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java:255)
	at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.put(ElasticsearchSinkTask.java:169)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)
	... 10 more
Caused by: java.net.SocketTimeoutException: Read timed out
	at java.net.SocketInputStream.socketRead0(Native Method)
	at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
	at java.net.SocketInputStream.read(SocketInputStream.java:171)
	at java.net.SocketInputStream.read(SocketInputStream.java:141)
	at org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
	at org.apache.http.impl.io.SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
	at org.apache.http.impl.io.SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:282)
	at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
	at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
	at org.apache.http.impl.io.AbstractMessageParser.parse(AbstractMessageParser.java:259)
	at org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
	at org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:165)
	at org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
	at org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
	at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
	at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:185)
	at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
	at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:111)
	at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:108)
	at io.searchbox.client.http.JestHttpClient.executeRequest(JestHttpClient.java:136)
	at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:70)
	at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:63)
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.indexExists(JestElasticsearchClient.java:281)
	... 14 more
```
Read timed out when trying to check if index exists
issue
created
levzem
username_1
@username_0 have you tried to increase that timeout configuration?
comment
created
VincEmond
username_0
Yes, we have a read timeout of 30 seconds.
comment
created
levzem
username_1
@username_0 how about more than 30s?
comment
created
VincEmond
username_0
I could try it. However, is this supposed to be this long? I am just wondering if such a timeout is normal for just a HEAD call, which instantly responds when curling directly Elastic.
comment
created
levzem
username_1
@username_0 another config that is good to try is the `connection.timeout.ms`

yeah it doesn't seem normal. The connector is successful in ingesting some data into ES for some time and then fails? Or does it immediately fail?
comment
created
VincEmond
username_0
It does succeed for some time. After a couple hours, all the tasks will die and I will see this exception.
The `connection.timeout.ms` is set to 20s in my case.
comment
created
levzem
username_1
hmm seems like could be an issue. Might be worth investigating. #189 might be a similar issue that you are experiencing. 

I see you are trying out the jar from #349. Did you get that to solve your issues?
comment
created
VincEmond
username_0
It indeeds look like #189 .

I tried using the configuration from #349, recompiled the jar and injected it in the docker container, but without success, still the same error, even after adding the maxConnectionIdleTimeout.
comment
created
levzem
username_1
Do you know if the connector is idle for some time i.e there is no data coming in for a while? or is it a continuous stream?
comment
created
VincEmond
username_0
There is no data coming in for a while. So I usually have a bunch of data at the beginning (since there is consumer lag because of the connectors dying earlier) that is all consumed by the connector. Afterwards, there could be some time before some data comes back in. The connectors usually die over the course of some data coming in little by little. We can say there is not always some data in the stream waiting to be consumed.
comment
created
wucheong
username_2
Hi @username_0 , I am having similar issue. The connector works at very beginning. And I leave it for couple hours, it will die with a first message of topics on next day.
comment
created
VincEmond
username_0
@username_1 do you have any insight on how we could fix this?
comment
created
VincEmond
username_0
@username_2 , are you able to consume some data when the connector first starts as well?
comment
created
wucheong
username_2
Yes, it did. And it just run pretty smooth if I keep producing data in the topic. Since there are no message at the midnight, it will die on next morning.
comment
created
levzem
username_1
@username_0 @username_2

this is beginning to smell like a bug. I don't have any suggestions for how to work around this at this time unfortunately (besides having continuous data, but that's not exactly feasible). I will try to find time to investigate the root cause, but if you guys want to dive into the code and find a fix, it would be a great contribution and I will happily review a PR. I suspect the fix will be relatively straightforward.
comment
created
levzem
username_1
@username_0 @username_2 

digging through the code and [documentation](https://hc.apache.org/httpcomponents-client-4.2.x/tutorial/html/connmgmt.html), it seems that if you set the read timeout (`read.timeout.ms`) to 0, this is will set it to an infinite timeout and should solve your problem. Please give this a go and let me know if you still experience the same issue.
comment
created
ftardif
username_3
I am experiencing the exact same problem as described above and I am not sure setting the read timeout to infinite will solve the issue (actually I believe this could simply hang forever the connector processing thread while reading from a dead connection). To what I experienced, it looks as if the TCP connection has actually been droped (maybe by inactivity on the server or by any network element in the path), but the jest client still gets a reference on it by borrowing  from, what I guess is, a TCP connection pool. This could explain why it only happens in scenarios where the connector resumes after some period of inactivity on the upstream topic. I did not have the chance yet to delve in the code to understand how jest configures the Apache http client, but I would expect the fix should imply some max timeout on TCP connection polling, or some TCP pre-check of connection before borrowing it from the pool.
comment
created
levzem
username_1
@username_3 if that was the case, I believe #349 would have solved the issue. 

It might be a case of however both #349 and setting an infinite timeout. A socket timeout for reads is usually because there is no communication from the server (because no data was sent) in the given timeout period and not a case of the server dropping the connection.
comment
created
wucheong
username_2
I have set the read.timeout.ms to 0, now it become flush timeout and unstable (randomly crash with flush timeout). And I am setting the flush timeout 0 now to have it a try.
comment
created
levzem
username_1
@username_2 actually, you want to set the `flush.timeout.ms` to a higher number, not lower. The connector needs more time to write the records, that's why it is crashing,
comment
created
wucheong
username_2
@username_0 @username_1 @username_3 Hello, I have check out the latest source code and add the ConnectionIdleTime parameter in JestElasticsearchClient and set to 60000ms. I have no more read timeout issue now. Let's add this as a parameter just like #349 and it should resolve the problem.
comment
created
wucheong
username_2
@username_1  I have created PR #390 for this. May I ask for your review?
comment
created
levzem
username_1
@username_2

did you try adjusting the `flush.timeout.ms`? We already have a PR out for idle time parameter. I would like to just confirm this is actually the fix because @username_0 already tried #349 and they said it did not solve the problem for them. 

Let's see if this fixes it for @username_0 and @username_3, then I will be happy to review it. But please do try the newest suggestions for the configs that I have sent first.
comment
created
ftardif
username_3
I am trying the connection pool idle time with 5.4.0 to double check if
that works.
comment
created
wucheong
username_2
I have tried adjusting `flush.timeout.ms` to bigger value and infinite value before. It didn't fix the issue.
comment
created
wucheong
username_2
comment
created
VincEmond
username_0
I started running my test yesterday, we will see how it goes. Up to now, it did not break. I will keep you posted tomorrow. What has changed is that I have the whole new version of kafka-connect instead of just updating the .jar, which might be why it is now working.
comment
created
VincEmond
username_0
So, @username_2 's version is now running for 2 whole days and the connectors are still alive. I believe this will be a good fix. @username_1 , do you think we could expect a dot release for this fix? Otherwise, I will look into copying the .jar directly in the docker container manually.
comment
created
levzem
username_1
Sounds great! @username_2 I will leave a review on your PR then.

@username_0 unfortunately right now the connector is tied to our CP releases, so you would only see this change in an upcoming CP 5.5 or CP 6.0 depending on which release it makes it into.
comment
created
VincEmond
username_0
Alright, thanks! I will simply copy the .jar manually until this is included in a release.
comment
created
levzem
username_1
closing because #390 was merged. Thanks everyone for your help!
comment
closed
levzem
username_1
issue
created
loveyraturi
username_4
I am still facing the issue since it was merged in #390 so i made a jar using master branch manually and used it .It works fine for sometime but produces the same issue afterwards I have set CONNECT_CONNECTIONS_MAX_IDLE_MS to 72000000  i think if the system remain idle for more than this time it will not work and throws the same issue. please let me know in case  there is any fix for this i have also set CONNECT_READ_TIMEOUT_MS  to 0 
i am using docker image v5.4.0 and created my custom image on top of that by replacing elastic search connector jar with my custom jar.

I will put in more information if needed but please let me know the fix to make it run infinitely
comment
created
jhansonhpe
username_5
@username_4 please check that you are using CONNECT_CONNECTION_MAX_IDLE_MS and not CONNECT_CONNECTIONS_MAX_IDLE_MS.  Note lack of plural on CONNECTION is what the code has at https://github.com/confluentinc/kafka-connect-elasticsearch/pull/390/commits/3cea2ce2a8883cf8191dc42e6b9082fa01d20ea2 and in the update at https://github.com/confluentinc/kafka-connect-elasticsearch/pull/390/commits/9e61375ae6ce279a7f9386a05e466c65aea819fc
comment
created
avgojzack
username_6
I see this fix in master, but not in any release branch yet. Am I missing something?
comment
created
levzem
username_1
@username_6 this was merged to master and unfortunately it didn't make it before the `5.5.x` branch was cut, so you can expect this to make into `6.0.x`. If you need the fix now, you can download and use the snapshot `master` version.
comment
created
lixueyu001
username_7
I think the 5.5. X version can greatly reduce the occurrence of this timeout problem, and also greatly improve the writing efficiency.
Compared with the previous version, index cache is added here。
[click me](https://github.com/confluentinc/kafka-connect-elasticsearch/blob/155d1c7a7698e376d4f307df41b24e806c68e984/src/main/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClient.java#L324)
comment