You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by SRIKANT MVS <sr...@gmail.com> on 2021/04/01 16:29:50 UTC
Kafka consumer wont recover after WakeupException
HI Team,
I am using camel-kafka (version: 3.2.0) for consuming messages.
Below is the flow
1. Kafka service consumes events from the topic
2. Make a call to the Server
3. When the server is not responding in 40ms, throw
ServerUnavailableException
4. Stopping Kafka consumer on the topic
5. Unsubscribing from the topic testTopicV1
6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
testTopicV1. Caused by: [org.apache.kafka.common.errors.WakeupException -
null]*
7.
*Also seen an InterruptedException with the message "Interrupted while
waiting for consumer heartbeat thread to close" Once this happens, then the
Kafka consumer never recovers and I have to start the consumer service
manually. Any idea why this happens and how to mitigate from this issue ?
2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(log:212)]
Error unsubscribing testTopicV1-Thread 0 from kafka topic testTopicV1.
Caused by: [org.apache.kafka.common.errors.WakeupException - null]
2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(doRun:397)]
Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
org.apache.kafka.common.errors.WakeupException: null at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
[camel-kafka-3.2.0.jar!/:3.2.0] at
java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source) [?:1.8.0_271] at
java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]Full
StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed delivery
for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1 caught:
com.example.exception.ServerUnavailableException: RestClientException
during rest call Message History (complete message history is disabled)
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
[from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
com.example.exception.ServerUnavailableException: RestClientException
during rest call at
com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
~[classes!/:?] at
org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
~[camel-bean-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
~[camel-bean-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
~[camel-bean-3.2.0.jar!/:3.2.0] at
org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
~[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
[camel-base-3.2.0.jar!/:3.2.0] at
org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
[camel-support-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
[camel-kafka-3.2.0.jar!/:3.2.0] at
java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source) [?:1.8.0_271] at
java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271] Caused
by: org.springframework.web.client.ResourceAccessException: I/O error on
POST request for "http://xxxxxxx": Read timed out; nested exception is
java.net.SocketTimeoutException: Read timed out at
org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused by:
java.net.SocketTimeoutException: Read timed out at
java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271] at
java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271] at
java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.io.SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.io.SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.io.AbstractMessageParser.parse(AbstractMessageParser.java:259)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
~[httpcore-4.4.13.jar!/:4.4.13] at
org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
~[httpclient-4.5.12.jar!/:4.5.12] at
org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more 2021-01-31
05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka consumer
on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel (camel-1)
thread #7 - KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
2.000 seconds for ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
4.001 seconds for ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
6.001 seconds for ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
8.007 seconds for ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
10.000 seconds for ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)] Forcing
shutdown of ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)] Forcing
shutdown of ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)] Shutdown of
ExecutorService:
org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
= 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated: false
took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel (camel-1)
thread #6 - KafkaConsumer[testTopicV1]
[camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
testTopicV1-Thread 0 from topic testTopicV1 2021-01-31 05:48:23.143+0000
INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
[clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
[Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
previously assigned partitions testTopicV1-13, testTopicV1-12,
testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31 05:48:23.142+0000
WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
[camel.component.kafka.KafkaConsumer(log:212)] Error during processing.
Exchange[ID-TestId-1611647443253-0-25]. Caused by:
[ServerUnavailableException - RestClientException during rest call]
ServerUnavailableException: RestClientException during rest call 2021-01-31
05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread | TestService
[clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
[Consumer clientId=consumer-TestService-5, groupId=TestService] Group
coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is unavailable or
invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN Camel
(camel-1) thread #6 - KafkaConsumer[testTopicV1]
[camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
[org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(doRun:397)]
Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
org.apache.kafka.common.errors.WakeupException: null at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
[camel-kafka-3.2.0.jar!/:3.2.0] at
java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source) [?:1.8.0_271] at
java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
[Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
previously assigned partitions testTopicV1-24, testTopicV1-21,
testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31 05:48:23.143+0000
WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
[clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
[Consumer clientId=consumer-TestService-5, groupId=TestService] Interrupted
while waiting for consumer heartbeat thread to close 2021-01-31
05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
KafkaConsumer[testTopicV1]
[clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
[Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
previously assigned partitions testTopicV1-13, testTopicV1-12,
testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31 05:48:23.144+0000
WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
[camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
[org.apache.kafka.common.errors.WakeupException - null]
org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
[Consumer clientId=consumer-TestService-6, groupId=TestService] Interrupted
while waiting for consumer heartbeat thread to close 2021-01-31
05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
KafkaConsumer[testTopicV1]
[clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
[Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
previously assigned partitions testTopicV1-24, testTopicV1-21,
testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31 05:48:23.144+0000
ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
[kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
clientId=consumer-TestService-5, groupId=TestService] Failed to close
coordinator org.apache.kafka.common.errors.InterruptException:
java.lang.InterruptedException at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
~[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
~[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
[kafka-clients-2.4.0.jar!/:?] at
org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
[kafka-clients-2.4.0.jar!/:?] at
org.apache.camel.util.IOHelper.close(IOHelper.java:341)
[camel-util-3.2.0.jar!/:3.2.0] at
org.apache.camel.util.IOHelper.close(IOHelper.java:403)
[camel-util-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
[camel-kafka-3.2.0.jar!/:3.2.0] at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
[camel-kafka-3.2.0.jar!/:3.2.0] at
java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
[?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source) [?:1.8.0_271] at
java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
[?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271] Caused
by: java.lang.InterruptedException ... 26 more *
*-RegardsSrikant Mantha*
Re: Kafka consumer wont recover after WakeupException
Posted by Claus Ibsen <cl...@gmail.com>.
Hi
Can you try with Camel 3.11 (on the way) and Kafka 2.8 clients.
On Thu, Jun 10, 2021 at 5:26 PM SRIKANT MVS <sr...@gmail.com> wrote:
>
> Hi Team,
> I upgraded the camel version to 3.9 and the kafka-client jar upgraded to
> 2.7.0. I could see that kafka consumers are able to resume from the
> WakeupException and InterruptedException. However, I see a new issue. I see
> the below error message that gets printed endlessly until the consumer
> service is restarted manually.
>
> 2021-06-09 23:02:09.228+0000 ERROR Camel (camel-1) thread #1 -
> KafkaConsumer[testTopicV1]
> [camel.processor.errorhandler.DefaultErrorHandler(log:205)] Failed delivery
> for (MessageId: *D3D1C394132DC66-0000000000000001 *on ExchangeId:
> D3D1C394132DC66-0000000000000001). Exhausted after delivery attempt: 1
> caught: java.lang.IllegalStateException: Consumer is not subscribed to any
> topics or assigned any partitions
>
> Message History (complete message history is disabled)
> ---------------------------------------------------------------------------------------------------------------------------------------
> RouteId ProcessorId Processor
> Elapsed (ms)
> [route1 ] [route1 ]
> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
> 7]
> [route1 ] [process3 ] [Processor@0x77c233af
> ] [ 0]
>
> Stacktrace
> ---------------------------------------------------------------------------------------------------------------------------------------
>
> java.lang.IllegalStateException: Consumer is not subscribed to any topics
> or assigned any partitions
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1218)
> ~[kafka-clients-2.7.0.jar!/:?]
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1161)
> ~[kafka-clients-2.7.0.jar!/:?]
> at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doPollRun(KafkaConsumer.java:345)
> [camel-kafka-3.9.0.jar!/:3.9.0]
> at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:273)
> [camel-kafka-3.9.0.jar!/:3.9.0]
> at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:238)
> [camel-kafka-3.9.0.jar!/:3.9.0]
> at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:?]
> at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
> [?:?]
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?]
> at java.lang.Thread.run(Thread.java:829) [?:?]
>
> Exception keeps appearing something like this:
> 1st occurance of the exception at time *23:00:34.195* prints
> D3D1C394132DC66-0000000000000000
> 2nd occurrence happens at *23:02:09.228* prints
> D3D1C394132DC66-000000000000000*1*
> 3rd occurrence happens at *23:02:14.235* prints
> D3D1C394132DC66-000000000000000*2*
> 4th occurrence happens at *23:02:19.243* prints
> D3D1C394132DC66-000000000000000*3*
>
> All these are printed by the same thread which was initially processing the
> original request before the exception started.
>
> However, when I checked the logs, the consumer threads are getting
> subscribed back to the topic.
>
> Not sure why this message is not processed and errors out endlessly. Any
> inputs are appreciated.
>
> -Regards
> Srikant Mantha
>
>
>
>
>
> On Thu, Apr 1, 2021 at 6:44 PM SRIKANT MVS <sr...@gmail.com> wrote:
>
> > Hi Andrea/Claus,
> > Thanks for the quick response. I am seeing this issue from camel 3.1.0,
> > also migration to camel 3.4.5 (java 8 support) and kafka client upgraded to
> > 2.5.0. Still the same issue is seen.
> >
> > Let me also give a try upgrading to 3.9.0.
> >
> > -Regards
> > Srikant Mantha
> >
> > On Thu, Apr 1, 2021 at 6:37 PM Andrea Cosentino <an...@gmail.com> wrote:
> >
> >> Please try with an LTS version 3.7.x. 3.2.0 was a development version.
> >>
> >> Il giorno gio 1 apr 2021 alle ore 18:30 SRIKANT MVS <
> >> srikant.mvs@gmail.com>
> >> ha scritto:
> >>
> >> > HI Team,
> >> >
> >> > I am using camel-kafka (version: 3.2.0) for consuming messages.
> >> > Below is the flow
> >> >
> >> > 1. Kafka service consumes events from the topic
> >> > 2. Make a call to the Server
> >> > 3. When the server is not responding in 40ms, throw
> >> > ServerUnavailableException
> >> > 4. Stopping Kafka consumer on the topic
> >> > 5. Unsubscribing from the topic testTopicV1
> >> > 6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
> >> > testTopicV1. Caused by:
> >> [org.apache.kafka.common.errors.WakeupException
> >> > -
> >> > null]*
> >> > 7.
> >> >
> >> >
> >> > *Also seen an InterruptedException with the message "Interrupted while
> >> > waiting for consumer heartbeat thread to close" Once this happens,
> >> then
> >> > the
> >> > Kafka consumer never recovers and I have to start the consumer
> >> service
> >> > manually. Any idea why this happens and how to mitigate from this
> >> issue
> >> > ?
> >> > 2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(log:212)]
> >> > Error unsubscribing testTopicV1-Thread 0 from kafka topic
> >> testTopicV1.
> >> > Caused by: [org.apache.kafka.common.errors.WakeupException - null]
> >> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(doRun:397)]
> >> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> >> > org.apache.kafka.common.errors.WakeupException: null at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> >> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> >> > [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> >> > Source) [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> >> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source)
> >> [?:1.8.0_271]Full
> >> > StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread
> >> #7
> >> > -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed
> >> > delivery
> >> > for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
> >> > ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1
> >> > caught:
> >> > com.example.exception.ServerUnavailableException: RestClientException
> >> > during rest call Message History (complete message history is
> >> disabled)
> >> >
> >> >
> >> ---------------------------------------------------------------------------------------------------------------------------------------
> >> > RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
> >> >
> >> >
> >> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
> >> > 40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
> >> >
> >> >
> >> ---------------------------------------------------------------------------------------------------------------------------------------
> >> > com.example.exception.ServerUnavailableException: RestClientException
> >> > during rest call at
> >> > com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
> >> > ~[classes!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
> >> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
> >> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
> >> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >> >
> >> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
> >> > ~[camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> > org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
> >> > [camel-base-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
> >> > [camel-support-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> >> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> >> > [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> >> > Source) [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> >> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> >> > Caused
> >> > by: org.springframework.web.client.ResourceAccessException: I/O
> >> error on
> >> > POST request for "http://xxxxxxx": Read timed out; nested exception
> >> is
> >> > java.net.SocketTimeoutException: Read timed out at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused
> >> by:
> >> > java.net.SocketTimeoutException: Read timed out at
> >> > java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271]
> >> at
> >> > java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271]
> >> at
> >> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> >> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> >> > org.apache.http.impl.io
> >> > .SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> > org.apache.http.impl.io
> >> > .SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> > org.apache.http.impl.io
> >> > .SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> >
> >> >
> >> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> > org.apache.http.impl.io
> >> > .AbstractMessageParser.parse(AbstractMessageParser.java:259)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> >
> >> >
> >> org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> >
> >> >
> >> org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> >
> >> >
> >> org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
> >> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >> >
> >> >
> >> org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> > org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
> >> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >> >
> >> >
> >> org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
> >> > ~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >> >
> >> >
> >> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> >> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more
> >> 2021-01-31
> >> > 05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka
> >> > consumer
> >> > on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel
> >> (camel-1)
> >> > thread #7 - KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> >> > Waited
> >> > 2.000 seconds for ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> >> > 05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> >> > Waited
> >> > 4.001 seconds for ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> >> > 05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> >> > Waited
> >> > 6.001 seconds for ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> >> > 05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> >> > Waited
> >> > 8.007 seconds for ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> >> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> >> > Waited
> >> > 10.000 seconds for ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> >> > 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)]
> >> Forcing
> >> > shutdown of ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
> >> > 2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)]
> >> Forcing
> >> > shutdown of ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
> >> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)]
> >> Shutdown
> >> > of
> >> > ExecutorService:
> >> >
> >> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> >> > [Shutting
> >> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> >> > tasks
> >> > = 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated:
> >> false
> >> > took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel
> >> (camel-1)
> >> > thread #6 - KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
> >> > testTopicV1-Thread 0 from topic testTopicV1 2021-01-31
> >> 05:48:23.143+0000
> >> > INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> >> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
> >> Revoke
> >> > previously assigned partitions testTopicV1-13, testTopicV1-12,
> >> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> >> > 05:48:23.142+0000
> >> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(log:212)] Error during
> >> processing.
> >> > Exchange[ID-TestId-1611647443253-0-25]. Caused by:
> >> > [ServerUnavailableException - RestClientException during rest call]
> >> > ServerUnavailableException: RestClientException during rest call
> >> > 2021-01-31
> >> > 05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread |
> >> TestService
> >> >
> >> >
> >> [clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
> >> > [Consumer clientId=consumer-TestService-5, groupId=TestService] Group
> >> > coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is
> >> > unavailable or
> >> > invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN
> >> > Camel
> >> > (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> >> > testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
> >> > [org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
> >> > 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(doRun:397)]
> >> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> >> > org.apache.kafka.common.errors.WakeupException: null at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> >> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> >> > [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> >> > Source) [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> >> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> >> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> >> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
> >> Revoke
> >> > previously assigned partitions testTopicV1-24, testTopicV1-21,
> >> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> >> > 05:48:23.143+0000
> >> > WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> >> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
> >> > Interrupted
> >> > while waiting for consumer heartbeat thread to close 2021-01-31
> >> > 05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
> >> > KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> >> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
> >> Revoke
> >> > previously assigned partitions testTopicV1-13, testTopicV1-12,
> >> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> >> > 05:48:23.144+0000
> >> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> >> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> >> > testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
> >> > [org.apache.kafka.common.errors.WakeupException - null]
> >> > org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
> >> > 05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> >> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
> >> > Interrupted
> >> > while waiting for consumer heartbeat thread to close 2021-01-31
> >> > 05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
> >> > KafkaConsumer[testTopicV1]
> >> >
> >> >
> >> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> >> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
> >> Revoke
> >> > previously assigned partitions testTopicV1-24, testTopicV1-21,
> >> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> >> > 05:48:23.144+0000
> >> > ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >> > [kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
> >> > clientId=consumer-TestService-5, groupId=TestService] Failed to close
> >> > coordinator org.apache.kafka.common.errors.InterruptException:
> >> > java.lang.InterruptedException at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> >> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
> >> > ~[kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
> >> > [kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
> >> > [kafka-clients-2.4.0.jar!/:?] at
> >> >
> >> >
> >> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
> >> > [kafka-clients-2.4.0.jar!/:?] at
> >> > org.apache.camel.util.IOHelper.close(IOHelper.java:341)
> >> > [camel-util-3.2.0.jar!/:3.2.0] at
> >> > org.apache.camel.util.IOHelper.close(IOHelper.java:403)
> >> > [camel-util-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> >
> >> >
> >> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> >> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> >> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> >> > [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> >> > Source) [?:1.8.0_271] at
> >> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> >> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> >> > Caused
> >> > by: java.lang.InterruptedException ... 26 more *
> >> >
> >> > *-RegardsSrikant Mantha*
> >> >
> >>
> >
--
Claus Ibsen
-----------------
http://davsclaus.com @davsclaus
Camel in Action 2: https://www.manning.com/ibsen2
Re: Kafka consumer wont recover after WakeupException
Posted by SRIKANT MVS <sr...@gmail.com>.
Hi Team,
I upgraded the camel version to 3.9 and the kafka-client jar upgraded to
2.7.0. I could see that kafka consumers are able to resume from the
WakeupException and InterruptedException. However, I see a new issue. I see
the below error message that gets printed endlessly until the consumer
service is restarted manually.
2021-06-09 23:02:09.228+0000 ERROR Camel (camel-1) thread #1 -
KafkaConsumer[testTopicV1]
[camel.processor.errorhandler.DefaultErrorHandler(log:205)] Failed delivery
for (MessageId: *D3D1C394132DC66-0000000000000001 *on ExchangeId:
D3D1C394132DC66-0000000000000001). Exhausted after delivery attempt: 1
caught: java.lang.IllegalStateException: Consumer is not subscribed to any
topics or assigned any partitions
Message History (complete message history is disabled)
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor
Elapsed (ms)
[route1 ] [route1 ]
[from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
7]
[route1 ] [process3 ] [Processor@0x77c233af
] [ 0]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
java.lang.IllegalStateException: Consumer is not subscribed to any topics
or assigned any partitions
at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1218)
~[kafka-clients-2.7.0.jar!/:?]
at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1161)
~[kafka-clients-2.7.0.jar!/:?]
at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doPollRun(KafkaConsumer.java:345)
[camel-kafka-3.9.0.jar!/:3.9.0]
at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:273)
[camel-kafka-3.9.0.jar!/:3.9.0]
at
org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:238)
[camel-kafka-3.9.0.jar!/:3.9.0]
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
[?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?]
at java.lang.Thread.run(Thread.java:829) [?:?]
Exception keeps appearing something like this:
1st occurance of the exception at time *23:00:34.195* prints
D3D1C394132DC66-0000000000000000
2nd occurrence happens at *23:02:09.228* prints
D3D1C394132DC66-000000000000000*1*
3rd occurrence happens at *23:02:14.235* prints
D3D1C394132DC66-000000000000000*2*
4th occurrence happens at *23:02:19.243* prints
D3D1C394132DC66-000000000000000*3*
All these are printed by the same thread which was initially processing the
original request before the exception started.
However, when I checked the logs, the consumer threads are getting
subscribed back to the topic.
Not sure why this message is not processed and errors out endlessly. Any
inputs are appreciated.
-Regards
Srikant Mantha
On Thu, Apr 1, 2021 at 6:44 PM SRIKANT MVS <sr...@gmail.com> wrote:
> Hi Andrea/Claus,
> Thanks for the quick response. I am seeing this issue from camel 3.1.0,
> also migration to camel 3.4.5 (java 8 support) and kafka client upgraded to
> 2.5.0. Still the same issue is seen.
>
> Let me also give a try upgrading to 3.9.0.
>
> -Regards
> Srikant Mantha
>
> On Thu, Apr 1, 2021 at 6:37 PM Andrea Cosentino <an...@gmail.com> wrote:
>
>> Please try with an LTS version 3.7.x. 3.2.0 was a development version.
>>
>> Il giorno gio 1 apr 2021 alle ore 18:30 SRIKANT MVS <
>> srikant.mvs@gmail.com>
>> ha scritto:
>>
>> > HI Team,
>> >
>> > I am using camel-kafka (version: 3.2.0) for consuming messages.
>> > Below is the flow
>> >
>> > 1. Kafka service consumes events from the topic
>> > 2. Make a call to the Server
>> > 3. When the server is not responding in 40ms, throw
>> > ServerUnavailableException
>> > 4. Stopping Kafka consumer on the topic
>> > 5. Unsubscribing from the topic testTopicV1
>> > 6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
>> > testTopicV1. Caused by:
>> [org.apache.kafka.common.errors.WakeupException
>> > -
>> > null]*
>> > 7.
>> >
>> >
>> > *Also seen an InterruptedException with the message "Interrupted while
>> > waiting for consumer heartbeat thread to close" Once this happens,
>> then
>> > the
>> > Kafka consumer never recovers and I have to start the consumer
>> service
>> > manually. Any idea why this happens and how to mitigate from this
>> issue
>> > ?
>> > 2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(log:212)]
>> > Error unsubscribing testTopicV1-Thread 0 from kafka topic
>> testTopicV1.
>> > Caused by: [org.apache.kafka.common.errors.WakeupException - null]
>> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(doRun:397)]
>> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
>> > org.apache.kafka.common.errors.WakeupException: null at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
>> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
>> > [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
>> > Source) [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
>> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source)
>> [?:1.8.0_271]Full
>> > StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread
>> #7
>> > -
>> > KafkaConsumer[testTopicV1]
>> > [camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed
>> > delivery
>> > for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
>> > ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1
>> > caught:
>> > com.example.exception.ServerUnavailableException: RestClientException
>> > during rest call Message History (complete message history is
>> disabled)
>> >
>> >
>> ---------------------------------------------------------------------------------------------------------------------------------------
>> > RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
>> >
>> >
>> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
>> > 40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
>> >
>> >
>> ---------------------------------------------------------------------------------------------------------------------------------------
>> > com.example.exception.ServerUnavailableException: RestClientException
>> > during rest call at
>> > com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
>> > ~[classes!/:?] at
>> >
>> >
>> org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
>> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
>> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
>> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
>> >
>> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
>> > ~[camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> > org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
>> > [camel-base-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
>> > [camel-support-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
>> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
>> > [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
>> > Source) [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
>> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
>> > Caused
>> > by: org.springframework.web.client.ResourceAccessException: I/O
>> error on
>> > POST request for "http://xxxxxxx": Read timed out; nested exception
>> is
>> > java.net.SocketTimeoutException: Read timed out at
>> >
>> >
>> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused
>> by:
>> > java.net.SocketTimeoutException: Read timed out at
>> > java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271]
>> at
>> > java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271]
>> at
>> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
>> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
>> > org.apache.http.impl.io
>> > .SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> > org.apache.http.impl.io
>> > .SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> > org.apache.http.impl.io
>> > .SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> >
>> >
>> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> > org.apache.http.impl.io
>> > .AbstractMessageParser.parse(AbstractMessageParser.java:259)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> >
>> >
>> org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> >
>> >
>> org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> >
>> >
>> org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
>> > ~[httpcore-4.4.13.jar!/:4.4.13] at
>> >
>> >
>> org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> > org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
>> > ~[httpclient-4.5.12.jar!/:4.5.12] at
>> >
>> >
>> org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
>> > ~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
>> >
>> >
>> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>> >
>> >
>> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
>> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more
>> 2021-01-31
>> > 05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka
>> > consumer
>> > on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel
>> (camel-1)
>> > thread #7 - KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
>> > Waited
>> > 2.000 seconds for ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
>> > 05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
>> > Waited
>> > 4.001 seconds for ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
>> > 05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
>> > Waited
>> > 6.001 seconds for ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
>> > 05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
>> > Waited
>> > 8.007 seconds for ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
>> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
>> > Waited
>> > 10.000 seconds for ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
>> > 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)]
>> Forcing
>> > shutdown of ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
>> > 2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)]
>> Forcing
>> > shutdown of ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
>> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)]
>> Shutdown
>> > of
>> > ExecutorService:
>> >
>> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
>> > [Shutting
>> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
>> > tasks
>> > = 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated:
>> false
>> > took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel
>> (camel-1)
>> > thread #6 - KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
>> > testTopicV1-Thread 0 from topic testTopicV1 2021-01-31
>> 05:48:23.143+0000
>> > INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
>> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
>> Revoke
>> > previously assigned partitions testTopicV1-13, testTopicV1-12,
>> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
>> > 05:48:23.142+0000
>> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(log:212)] Error during
>> processing.
>> > Exchange[ID-TestId-1611647443253-0-25]. Caused by:
>> > [ServerUnavailableException - RestClientException during rest call]
>> > ServerUnavailableException: RestClientException during rest call
>> > 2021-01-31
>> > 05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread |
>> TestService
>> >
>> >
>> [clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
>> > [Consumer clientId=consumer-TestService-5, groupId=TestService] Group
>> > coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is
>> > unavailable or
>> > invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN
>> > Camel
>> > (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
>> > testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
>> > [org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
>> > 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(doRun:397)]
>> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
>> > org.apache.kafka.common.errors.WakeupException: null at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
>> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
>> > [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
>> > Source) [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
>> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
>> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
>> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
>> Revoke
>> > previously assigned partitions testTopicV1-24, testTopicV1-21,
>> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
>> > 05:48:23.143+0000
>> > WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
>> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
>> > Interrupted
>> > while waiting for consumer heartbeat thread to close 2021-01-31
>> > 05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
>> > KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
>> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
>> Revoke
>> > previously assigned partitions testTopicV1-13, testTopicV1-12,
>> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
>> > 05:48:23.144+0000
>> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
>> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
>> > testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
>> > [org.apache.kafka.common.errors.WakeupException - null]
>> > org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
>> > 05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
>> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
>> > Interrupted
>> > while waiting for consumer heartbeat thread to close 2021-01-31
>> > 05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
>> > KafkaConsumer[testTopicV1]
>> >
>> >
>> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
>> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
>> Revoke
>> > previously assigned partitions testTopicV1-24, testTopicV1-21,
>> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
>> > 05:48:23.144+0000
>> > ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>> > [kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
>> > clientId=consumer-TestService-5, groupId=TestService] Failed to close
>> > coordinator org.apache.kafka.common.errors.InterruptException:
>> > java.lang.InterruptedException at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
>> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
>> > ~[kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
>> > [kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
>> > [kafka-clients-2.4.0.jar!/:?] at
>> >
>> >
>> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
>> > [kafka-clients-2.4.0.jar!/:?] at
>> > org.apache.camel.util.IOHelper.close(IOHelper.java:341)
>> > [camel-util-3.2.0.jar!/:3.2.0] at
>> > org.apache.camel.util.IOHelper.close(IOHelper.java:403)
>> > [camel-util-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> >
>> >
>> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
>> > [camel-kafka-3.2.0.jar!/:3.2.0] at
>> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
>> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
>> > [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
>> > Source) [?:1.8.0_271] at
>> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
>> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
>> > Caused
>> > by: java.lang.InterruptedException ... 26 more *
>> >
>> > *-RegardsSrikant Mantha*
>> >
>>
>
Re: Kafka consumer wont recover after WakeupException
Posted by SRIKANT MVS <sr...@gmail.com>.
Hi Andrea/Claus,
Thanks for the quick response. I am seeing this issue from camel 3.1.0,
also migration to camel 3.4.5 (java 8 support) and kafka client upgraded to
2.5.0. Still the same issue is seen.
Let me also give a try upgrading to 3.9.0.
-Regards
Srikant Mantha
On Thu, Apr 1, 2021 at 6:37 PM Andrea Cosentino <an...@gmail.com> wrote:
> Please try with an LTS version 3.7.x. 3.2.0 was a development version.
>
> Il giorno gio 1 apr 2021 alle ore 18:30 SRIKANT MVS <srikant.mvs@gmail.com
> >
> ha scritto:
>
> > HI Team,
> >
> > I am using camel-kafka (version: 3.2.0) for consuming messages.
> > Below is the flow
> >
> > 1. Kafka service consumes events from the topic
> > 2. Make a call to the Server
> > 3. When the server is not responding in 40ms, throw
> > ServerUnavailableException
> > 4. Stopping Kafka consumer on the topic
> > 5. Unsubscribing from the topic testTopicV1
> > 6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
> > testTopicV1. Caused by:
> [org.apache.kafka.common.errors.WakeupException
> > -
> > null]*
> > 7.
> >
> >
> > *Also seen an InterruptedException with the message "Interrupted while
> > waiting for consumer heartbeat thread to close" Once this happens,
> then
> > the
> > Kafka consumer never recovers and I have to start the consumer service
> > manually. Any idea why this happens and how to mitigate from this
> issue
> > ?
> > 2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
> > KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(log:212)]
> > Error unsubscribing testTopicV1-Thread 0 from kafka topic testTopicV1.
> > Caused by: [org.apache.kafka.common.errors.WakeupException - null]
> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(doRun:397)]
> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> > org.apache.kafka.common.errors.WakeupException: null at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> > [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> > Source) [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source)
> [?:1.8.0_271]Full
> > StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread
> #7
> > -
> > KafkaConsumer[testTopicV1]
> > [camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed
> > delivery
> > for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
> > ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1
> > caught:
> > com.example.exception.ServerUnavailableException: RestClientException
> > during rest call Message History (complete message history is
> disabled)
> >
> >
> ---------------------------------------------------------------------------------------------------------------------------------------
> > RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
> >
> >
> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
> > 40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
> >
> >
> ---------------------------------------------------------------------------------------------------------------------------------------
> > com.example.exception.ServerUnavailableException: RestClientException
> > during rest call at
> > com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
> > ~[classes!/:?] at
> >
> >
> org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
> > ~[camel-bean-3.2.0.jar!/:3.2.0] at
> >
> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
> > ~[camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> > org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
> > [camel-base-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
> > [camel-support-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> > [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> > Source) [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> > Caused
> > by: org.springframework.web.client.ResourceAccessException: I/O error
> on
> > POST request for "http://xxxxxxx": Read timed out; nested exception
> is
> > java.net.SocketTimeoutException: Read timed out at
> >
> >
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused
> by:
> > java.net.SocketTimeoutException: Read timed out at
> > java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271]
> at
> > java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271]
> at
> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> > java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> > org.apache.http.impl.io
> > .SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> > org.apache.http.impl.io
> > .SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> > org.apache.http.impl.io
> > .SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >
> >
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> > org.apache.http.impl.io
> > .AbstractMessageParser.parse(AbstractMessageParser.java:259)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >
> >
> org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >
> >
> org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >
> >
> org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
> > ~[httpcore-4.4.13.jar!/:4.4.13] at
> >
> >
> org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> > org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
> > ~[httpclient-4.5.12.jar!/:4.5.12] at
> >
> >
> org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
> > ~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
> >
> >
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> >
> >
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> > ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more
> 2021-01-31
> > 05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka
> > consumer
> > on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel
> (camel-1)
> > thread #7 - KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> > Waited
> > 2.000 seconds for ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> > 05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> > Waited
> > 4.001 seconds for ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> > 05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> > Waited
> > 6.001 seconds for ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> > 05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> > Waited
> > 8.007 seconds for ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> > Waited
> > 10.000 seconds for ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> > 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)] Forcing
> > shutdown of ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
> > 2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)] Forcing
> > shutdown of ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
> > 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)]
> Shutdown
> > of
> > ExecutorService:
> > org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> > [Shutting
> > down, pool size = 2, active threads = 2, queued tasks = 0, completed
> > tasks
> > = 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated:
> false
> > took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel
> (camel-1)
> > thread #6 - KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
> > testTopicV1-Thread 0 from topic testTopicV1 2021-01-31
> 05:48:23.143+0000
> > INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> > [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> > previously assigned partitions testTopicV1-13, testTopicV1-12,
> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> > 05:48:23.142+0000
> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(log:212)] Error during
> processing.
> > Exchange[ID-TestId-1611647443253-0-25]. Caused by:
> > [ServerUnavailableException - RestClientException during rest call]
> > ServerUnavailableException: RestClientException during rest call
> > 2021-01-31
> > 05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread |
> TestService
> >
> >
> [clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
> > [Consumer clientId=consumer-TestService-5, groupId=TestService] Group
> > coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is
> > unavailable or
> > invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN
> > Camel
> > (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> > testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
> > [org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
> > 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(doRun:397)]
> > Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> > org.apache.kafka.common.errors.WakeupException: null at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> > [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> > Source) [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> > 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> > [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> > previously assigned partitions testTopicV1-24, testTopicV1-21,
> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> > 05:48:23.143+0000
> > WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> > [Consumer clientId=consumer-TestService-5, groupId=TestService]
> > Interrupted
> > while waiting for consumer heartbeat thread to close 2021-01-31
> > 05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
> > KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> > [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> > previously assigned partitions testTopicV1-13, testTopicV1-12,
> > testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> > 05:48:23.144+0000
> > WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> > [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> > testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
> > [org.apache.kafka.common.errors.WakeupException - null]
> > org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
> > 05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> > [Consumer clientId=consumer-TestService-6, groupId=TestService]
> > Interrupted
> > while waiting for consumer heartbeat thread to close 2021-01-31
> > 05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
> > KafkaConsumer[testTopicV1]
> >
> >
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> > [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> > previously assigned partitions testTopicV1-24, testTopicV1-21,
> > testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> > 05:48:23.144+0000
> > ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> > [kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
> > clientId=consumer-TestService-5, groupId=TestService] Failed to close
> > coordinator org.apache.kafka.common.errors.InterruptException:
> > java.lang.InterruptedException at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> > ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
> > ~[kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
> > [kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
> > [kafka-clients-2.4.0.jar!/:?] at
> >
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
> > [kafka-clients-2.4.0.jar!/:?] at
> > org.apache.camel.util.IOHelper.close(IOHelper.java:341)
> > [camel-util-3.2.0.jar!/:3.2.0] at
> > org.apache.camel.util.IOHelper.close(IOHelper.java:403)
> > [camel-util-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> >
> >
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> > [camel-kafka-3.2.0.jar!/:3.2.0] at
> > java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> > [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> > [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> > Source) [?:1.8.0_271] at
> > java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> > [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> > Caused
> > by: java.lang.InterruptedException ... 26 more *
> >
> > *-RegardsSrikant Mantha*
> >
>
Re: Kafka consumer wont recover after WakeupException
Posted by Andrea Cosentino <an...@gmail.com>.
Please try with an LTS version 3.7.x. 3.2.0 was a development version.
Il giorno gio 1 apr 2021 alle ore 18:30 SRIKANT MVS <sr...@gmail.com>
ha scritto:
> HI Team,
>
> I am using camel-kafka (version: 3.2.0) for consuming messages.
> Below is the flow
>
> 1. Kafka service consumes events from the topic
> 2. Make a call to the Server
> 3. When the server is not responding in 40ms, throw
> ServerUnavailableException
> 4. Stopping Kafka consumer on the topic
> 5. Unsubscribing from the topic testTopicV1
> 6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
> testTopicV1. Caused by: [org.apache.kafka.common.errors.WakeupException
> -
> null]*
> 7.
>
>
> *Also seen an InterruptedException with the message "Interrupted while
> waiting for consumer heartbeat thread to close" Once this happens, then
> the
> Kafka consumer never recovers and I have to start the consumer service
> manually. Any idea why this happens and how to mitigate from this issue
> ?
> 2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
> KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)]
> Error unsubscribing testTopicV1-Thread 0 from kafka topic testTopicV1.
> Caused by: [org.apache.kafka.common.errors.WakeupException - null]
> 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doRun:397)]
> Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> org.apache.kafka.common.errors.WakeupException: null at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]Full
> StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread #7
> -
> KafkaConsumer[testTopicV1]
> [camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed
> delivery
> for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
> ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1
> caught:
> com.example.exception.ServerUnavailableException: RestClientException
> during rest call Message History (complete message history is disabled)
>
> ---------------------------------------------------------------------------------------------------------------------------------------
> RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
>
> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
> 40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
>
> ---------------------------------------------------------------------------------------------------------------------------------------
> com.example.exception.ServerUnavailableException: RestClientException
> during rest call at
> com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
> ~[classes!/:?] at
>
> org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
> ~[camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
> [camel-base-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
> [camel-support-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> Caused
> by: org.springframework.web.client.ResourceAccessException: I/O error on
> POST request for "http://xxxxxxx": Read timed out; nested exception is
> java.net.SocketTimeoutException: Read timed out at
>
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused by:
> java.net.SocketTimeoutException: Read timed out at
> java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271] at
> java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271] at
> java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> org.apache.http.impl.io
> .SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.io
> .SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.io
> .SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
>
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.io
> .AbstractMessageParser.parse(AbstractMessageParser.java:259)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
>
> org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
>
> org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
>
> org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
>
> org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
>
> org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
> ~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
>
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
>
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more 2021-01-31
> 05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka
> consumer
> on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel (camel-1)
> thread #7 - KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> Waited
> 2.000 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> Waited
> 4.001 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> Waited
> 6.001 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> Waited
> 8.007 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)]
> Waited
> 10.000 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)] Forcing
> shutdown of ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
> 2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)] Forcing
> shutdown of ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
> 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)] Shutdown
> of
> ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0
> [Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed
> tasks
> = 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated: false
> took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel (camel-1)
> thread #6 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
> testTopicV1-Thread 0 from topic testTopicV1 2021-01-31 05:48:23.143+0000
> INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-13, testTopicV1-12,
> testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> 05:48:23.142+0000
> WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error during processing.
> Exchange[ID-TestId-1611647443253-0-25]. Caused by:
> [ServerUnavailableException - RestClientException during rest call]
> ServerUnavailableException: RestClientException during rest call
> 2021-01-31
> 05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread | TestService
>
> [clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Group
> coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is
> unavailable or
> invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN
> Camel
> (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
> [org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
> 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doRun:397)]
> Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> org.apache.kafka.common.errors.WakeupException: null at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-24, testTopicV1-21,
> testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> 05:48:23.143+0000
> WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService]
> Interrupted
> while waiting for consumer heartbeat thread to close 2021-01-31
> 05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
> KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-13, testTopicV1-12,
> testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31
> 05:48:23.144+0000
> WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
> [org.apache.kafka.common.errors.WakeupException - null]
> org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
> 05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService]
> Interrupted
> while waiting for consumer heartbeat thread to close 2021-01-31
> 05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
>
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-24, testTopicV1-21,
> testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31
> 05:48:23.144+0000
> ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
> clientId=consumer-TestService-5, groupId=TestService] Failed to close
> coordinator org.apache.kafka.common.errors.InterruptException:
> java.lang.InterruptedException at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
> ~[kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
> [kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
> [kafka-clients-2.4.0.jar!/:?] at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
> [kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.util.IOHelper.close(IOHelper.java:341)
> [camel-util-3.2.0.jar!/:3.2.0] at
> org.apache.camel.util.IOHelper.close(IOHelper.java:403)
> [camel-util-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
>
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> Caused
> by: java.lang.InterruptedException ... 26 more *
>
> *-RegardsSrikant Mantha*
>
Re: Kafka consumer wont recover after WakeupException
Posted by Claus Ibsen <cl...@gmail.com>.
Hi
That is an old Camel version. Upgrade to a newer version like 3.9.0
On Thu, Apr 1, 2021 at 6:30 PM SRIKANT MVS <sr...@gmail.com> wrote:
>
> HI Team,
>
> I am using camel-kafka (version: 3.2.0) for consuming messages.
> Below is the flow
>
> 1. Kafka service consumes events from the topic
> 2. Make a call to the Server
> 3. When the server is not responding in 40ms, throw
> ServerUnavailableException
> 4. Stopping Kafka consumer on the topic
> 5. Unsubscribing from the topic testTopicV1
> 6. *Error unsubscribing testTopicV1-Thread 0 from kafka topic
> testTopicV1. Caused by: [org.apache.kafka.common.errors.WakeupException -
> null]*
> 7.
>
>
> *Also seen an InterruptedException with the message "Interrupted while
> waiting for consumer heartbeat thread to close" Once this happens, then the
> Kafka consumer never recovers and I have to start the consumer service
> manually. Any idea why this happens and how to mitigate from this issue ?
> 2021-01-31 05:48:23.143+0000 WARN Camel (camel-1) thread #6 -
> KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(log:212)]
> Error unsubscribing testTopicV1-Thread 0 from kafka topic testTopicV1.
> Caused by: [org.apache.kafka.common.errors.WakeupException - null]
> 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(doRun:397)]
> Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> org.apache.kafka.common.errors.WakeupException: null at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]Full
> StackTrace 2021-01-31 05:48:13.138+0000 ERROR Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.processor.errorhandler.DefaultErrorHandler(log:203)] Failed delivery
> for (MessageId: ID-TestId-1611647443253-0-25 on ExchangeId:
> ID-TestId-1611647443253-0-25). Exhausted after delivery attempt: 1 caught:
> com.example.exception.ServerUnavailableException: RestClientException
> during rest call Message History (complete message history is disabled)
> ---------------------------------------------------------------------------------------------------------------------------------------
> RouteId ProcessorId Processor Elapsed (ms) [route1 ] [route1 ]
> [from[kafka://testTopicV1?allowManualCommit=True&autoCommitEnable=False] [
> 40027] ... [route1 ] [to1 ] [bean:kafkaProcessor ] [ 0] Stacktrace
> ---------------------------------------------------------------------------------------------------------------------------------------
> com.example.exception.ServerUnavailableException: RestClientException
> during rest call at
> com.example.processor.KafkaProcessor.process(KafkaProcessor.java:46)
> ~[classes!/:?] at
> org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:117)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:56)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41)
> ~[camel-bean-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.SendProcessor.process(SendProcessor.java:166)
> ~[camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:286)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.impl.engine.DefaultAsyncProcessorAwaitManager.process(DefaultAsyncProcessorAwaitManager.java:83)
> [camel-base-3.2.0.jar!/:3.2.0] at
> org.apache.camel.support.AsyncProcessorSupport.process(AsyncProcessorSupport.java:40)
> [camel-support-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:338)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271] Caused
> by: org.springframework.web.client.ResourceAccessException: I/O error on
> POST request for "http://xxxxxxx": Read timed out; nested exception is
> java.net.SocketTimeoutException: Read timed out at
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:746)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more Caused by:
> java.net.SocketTimeoutException: Read timed out at
> java.net.SocketInputStream.socketRead0(Native Method) ~[?:1.8.0_271] at
> java.net.SocketInputStream.socketRead(Unknown Source) ~[?:1.8.0_271] at
> java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> java.net.SocketInputStream.read(Unknown Source) ~[?:1.8.0_271] at
> org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.io.SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.io.SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:280)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.io.AbstractMessageParser.parse(AbstractMessageParser.java:259)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:157)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
> ~[httpcore-4.4.13.jar!/:4.4.13] at
> org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
> ~[httpclient-4.5.12.jar!/:4.5.12] at
> org.springframework.http.client.HttpComponentsClientHttpRequest.executeInternal(HttpComponentsClientHttpRequest.java:87)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:109)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.boot.actuate.metrics.web.client.MetricsClientHttpRequestInterceptor.intercept(MetricsClientHttpRequestInterceptor.java:95)
> ~[spring-boot-actuator-2.2.11.RELEASE.jar!/:2.2.11.RELEASE] at
> org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:77)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:737)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.web.client.RestTemplate.execute(RestTemplate.java:672)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] at
> org.springframework.web.client.RestTemplate.postForEntity(RestTemplate.java:447)
> ~[spring-web-5.2.10.RELEASE.jar!/:5.2.10.RELEASE] ... 21 more 2021-01-31
> 05:48:13.139+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doStop:141)] Stopping Kafka consumer
> on topic: testTopicV1 2021-01-31 05:48:15.140+0000 INFO Camel (camel-1)
> thread #7 - KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
> 2.000 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:17.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
> 4.001 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:19.142+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
> 6.001 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:21.147+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
> 8.007 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(awaitTermination:405)] Waited
> 10.000 seconds for ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] to terminate... 2021-01-31
> 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:305)] Forcing
> shutdown of ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] due first await termination elapsed.
> 2021-01-31 05:48:23.141+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:314)] Forcing
> shutdown of ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] due interrupted. 2021-01-31
> 05:48:23.141+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [camel.impl.engine.BaseExecutorServiceManager(doShutdown:322)] Shutdown of
> ExecutorService:
> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@35c175c0[Shutting
> down, pool size = 2, active threads = 2, queued tasks = 0, completed tasks
> = 0][KafkaConsumer[testTopicV1]] is shutdown: true and terminated: false
> took: 10.001 seconds. 2021-01-31 05:48:23.142+0000 INFO Camel (camel-1)
> thread #6 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(doRun:397)] Unsubscribing
> testTopicV1-Thread 0 from topic testTopicV1 2021-01-31 05:48:23.143+0000
> INFO Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-13, testTopicV1-12,
> testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31 05:48:23.142+0000
> WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error during processing.
> Exchange[ID-TestId-1611647443253-0-25]. Caused by:
> [ServerUnavailableException - RestClientException during rest call]
> ServerUnavailableException: RestClientException during rest call 2021-01-31
> 05:48:23.143+0000 INFO kafka-coordinator-heartbeat-thread | TestService
> [clients.consumer.internals.AbstractCoordinator(markCoordinatorUnknown:808)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Group
> coordinator KafkaBroker3:9093 (id: 2147482641 rack: null) is unavailable or
> invalid, will attempt rediscovery 2021-01-31 05:48:23.143+0000 WARN Camel
> (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> testTopicV1-Thread 0 from kafka topic testTopicV1. Caused by:
> [org.apache.kafka.common.errors.WakeupException - null] 2021-01-31
> 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1] [camel.component.kafka.KafkaConsumer(doRun:397)]
> Unsubscribing testTopicV1-Thread 1 from topic testTopicV1
> org.apache.kafka.common.errors.WakeupException: null at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:511)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:275)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:937)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:400)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
> 2021-01-31 05:48:23.143+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-24, testTopicV1-21,
> testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31 05:48:23.143+0000
> WARN Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Interrupted
> while waiting for consumer heartbeat thread to close 2021-01-31
> 05:48:23.144+0000 INFO Camel (camel-1) thread #6 -
> KafkaConsumer[testTopicV1]
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-5, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-13, testTopicV1-12,
> testTopicV1-14, testTopicV1-11, testTopicV1-10 2021-01-31 05:48:23.144+0000
> WARN Camel (camel-1) thread #7 - KafkaConsumer[testTopicV1]
> [camel.component.kafka.KafkaConsumer(log:212)] Error unsubscribing
> testTopicV1-Thread 1 from kafka topic testTopicV1. Caused by:
> [org.apache.kafka.common.errors.WakeupException - null]
> org.apache.kafka.common.errors.WakeupException: null .... 2021-01-31
> 05:48:23.144+0000 WARN Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [clients.consumer.internals.AbstractCoordinator(closeHeartbeatThread:367)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService] Interrupted
> while waiting for consumer heartbeat thread to close 2021-01-31
> 05:48:23.144+0000 INFO Camel (camel-1) thread #7 -
> KafkaConsumer[testTopicV1]
> [clients.consumer.internals.ConsumerCoordinator(invokePartitionsRevoked:286)]
> [Consumer clientId=consumer-TestService-6, groupId=TestService] Revoke
> previously assigned partitions testTopicV1-24, testTopicV1-21,
> testTopicV1-20, testTopicV1-23, testTopicV1-22 2021-01-31 05:48:23.144+0000
> ERROR Camel (camel-1) thread #6 - KafkaConsumer[testTopicV1]
> [kafka.clients.consumer.KafkaConsumer(close:2297)] [Consumer
> clientId=consumer-TestService-5, groupId=TestService] Failed to close
> coordinator org.apache.kafka.common.errors.InterruptException:
> java.lang.InterruptedException at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:212)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:932)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1473)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1431)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.commitOffset(KafkaConsumer.java:432)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.onPartitionsRevoked(KafkaConsumer.java:455)
> ~[camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:887)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:832)
> ~[kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2294)
> [kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2261)
> [kafka-clients-2.4.0.jar!/:?] at
> org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2211)
> [kafka-clients-2.4.0.jar!/:?] at
> org.apache.camel.util.IOHelper.close(IOHelper.java:341)
> [camel-util-3.2.0.jar!/:3.2.0] at
> org.apache.camel.util.IOHelper.close(IOHelper.java:403)
> [camel-util-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.doRun(KafkaConsumer.java:419)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> org.apache.camel.component.kafka.KafkaConsumer$KafkaFetchRecords.run(KafkaConsumer.java:214)
> [camel-kafka-3.2.0.jar!/:3.2.0] at
> java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.FutureTask.run(Unknown Source)
> [?:1.8.0_271] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source) [?:1.8.0_271] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> [?:1.8.0_271] at java.lang.Thread.run(Unknown Source) [?:1.8.0_271] Caused
> by: java.lang.InterruptedException ... 26 more *
>
> *-RegardsSrikant Mantha*
--
Claus Ibsen
-----------------
http://davsclaus.com @davsclaus
Camel in Action 2: https://www.manning.com/ibsen2