search

wso2 / product-micro-integrator

The cloud-native configuration driven runtime that helps developers implement composite microservices.
https://wso2.com/integration/
Apache License 2.0
204 stars 221 forks source link

[Bug]: Reconnecting to kafka servers must have a suspend on retry #3546

Open anupama-pathirage opened 3 months ago

anupama-pathirage commented 3 months ago

Description

Moving the issue from https://github.com/wso2-extensions/esb-inbound-kafka/issues/19

image

Description: If Inbound cannot connect to Kafka servers, it will keep trying with warnings. This will cause the server to overload eventually. There should be a suspend option of the inbound after retrying a few times, similar to JMS or RabbitMQ transports.

Affected Product Version: 1.0.3

OS, DB, other environment details and versions: N/A

Steps to reproduce: Start ESB with a Kafka Inbound Enpoint with servers not working.

Refer following logs : 

TID: [] [] [2018-05-12 15:51:04,714] ERROR {org.wso2.carbon.ntask.core.impl.TaskQuartzJobAdapter} -  Error in executing task: Failed to construct kafka consumer {org.wso2.carbon.ntask.core.impl.TaskQuartzJobAdapter}
org.apache.synapse.SynapseException: Failed to construct kafka consumer
        at org.wso2.carbon.inbound.kafka.KafkaMessageConsumer.consumeKafkaRecords(KafkaMessageConsumer.java:294)
        at org.wso2.carbon.inbound.kafka.KafkaMessageConsumer.poll(KafkaMessageConsumer.java:423)
        at org.wso2.carbon.inbound.endpoint.protocol.generic.GenericTask.taskExecute(GenericTask.java:41)
        at org.wso2.carbon.inbound.endpoint.common.InboundTask.execute(InboundTask.java:51)
        at org.wso2.carbon.mediation.ntask.NTaskAdapter.execute(NTaskAdapter.java:98)
        at org.wso2.carbon.ntask.core.impl.TaskQuartzJobAdapter.execute(TaskQuartzJobAdapter.java:67)
        at org.quartz.core.JobRunShell.run(JobRunShell.java:213)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:765)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:633)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:615)
        at org.wso2.carbon.inbound.kafka.KafkaMessageConsumer.consumeKafkaRecords(KafkaMessageConsumer.java:292)
        ... 11 more

Steps to Reproduce

No response

Affected Version(s)

No response

OS, DB, other environment details and versions

No response

Related issue(s) (optional)

No response

Suggested label(s) (optional)

No response

Suggested assignee(s) (optional)

No response