Multiple adapter processes polling the same Queue

[ibm-messaging]

Multiple adapter processes polling the same Queue

[AndrewJSchofield]

Just run the connector in distributed mode. Rather than using connect-standalone.sh, use two or more instances of connect-distributed.sh. Then use the Kafka Connect REST API to start the connector and it will be automatically made highly available among the distributed workers. No change to the connector required.

[SAshish19]

Thanks Andrew.
I tried to run the connector in distributed mode and it seems to be running fine. I see the java process listening on the default port of 8083 .However, I am getting HTTP 400 error while trying to post a curl request :

[root@ip-172-31-40-55 kafka_2.11-2.0.0]# curl -X POST -H "Content-Type: application/json" --data '{"name": "mq-source", "config": {"mq.record.builder":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector", "tasks.max":"1", "mq.queue":"KAFKA", "mq.queue.manager":"XXXXXX" }}' http://localhost:8083/connectors
{"error_code":400,"message":"Connector config {mq.record.builder=com.ibm.eventstreams.connect.mqsource.MQSourceConnector, tasks.max=1, mq.queue=KAFKA, mq.queue.manager=XXXXX, name=mq-source} contains no connector type"}

Am i missing anything on my Curl request ?

Regards

[AndrewJSchofield]

You have typed mq.record.builder but it should be connector.class.

[SAshish19]

Hi Andrew,

I replaced "mq.record.builder" with "connector.class" but got the below error (looks like mq.record.builder is a mandatory element) :

curl -X POST -H "Content-Type: application/json" --data '{"name": "mq-source", "config": {"connector.class":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector", "tasks.max":"1", "mq.queue":"KAFKA", "mq.queue.manager":"FMQSOCQM02" }}' http://localhost:8083/connectors

{"error_code":400,"message":"Connector configuration is invalid and contains the following 1 error(s):\nMissing required configuration "mq.record.builder" which has no default value.\nYou can also find the above list of errors at the endpoint /{connectorType}/config/validate"}

I added both "connector.class" and "mq.record.builder" with the class name and got the below response without any error code :

curl -X POST -H "Content-Type: application/json" --data '{"name": "mq-source", "config": {"connector.class":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector", "tasks.max":"1", "mq.queue":"KAFKA", "mq.queue.manager":"FMQSOCQM02", "mq.record.builder":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector" }}' http://localhost:8083/connectors
{"name":"mq-source","config":{"connector.class":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector","tasks.max":"1","mq.queue":"KAFKA","mq.queue.manager":"FMQSOCQM02","mq.record.builder":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector","name":"mq-source"},"tasks":[],"type":null}

But immediately the connector in distributed mode started getting the JMS exception (and now for some strange reason every time I am starting the connector in distributed mode I keep
getting the same error) :

ERROR WorkerSourceTask{id=mq-source-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
org.apache.kafka.connect.errors.ConnectException: com.ibm.msg.client.jms.DetailedJMSException: JMSCC0005: The specified value '' is not allowed for 'XMSC_WMQ_CHANNEL'.
The given value is not allowed for the property specified.
Change the value to a value that is supported for the property.
at com.ibm.eventstreams.connect.mqsource.JMSReader.configure(JMSReader.java:164)
at com.ibm.eventstreams.connect.mqsource.MQSourceTask.start(MQSourceTask.java:81)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:198)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
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: com.ibm.msg.client.jms.DetailedJMSException: JMSCC0005: The specified value '' is not allowed for 'XMSC_WMQ_CHANNEL'.

The error : com.ibm.msg.client.jms.DetailedJMSException: JMSCC0005: The specified value '' is not allowed for 'XMSC_WMQ_CHANNEL'. is a bit confusing because the mq-source.properties file is correctly set with the channel name.

Regards

[SAshish19]

Hi Andrew,

I was able to get rid of "JMSCC0005: The specified value '' is not allowed for 'XMSC_WMQ_CHANNEL'." error. It seems we have to pass the mq.channel.name config as well in the REST command. Below is what the updated command looks like :

curl -X POST -H "Content-Type: application/json" --data '{"name": "mq-source1", "config": {"connector.class":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector", "tasks.max":"1", "mq.queue":"KAFKA", "mq.queue.manager":"KAFKA_QM", "mq.channel.name":"KAFKA_DIS_SVRCONN", "mq.record.builder":"com.ibm.eventstreams.connect.mqsource.MQSourceConnector" }}' http://localhost:8085/connectors

However, I still got below error :

ERROR WorkerSourceTask{id=mq-source1-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
java.lang.ClassCastException: class com.ibm.eventstreams.connect.mqsource.MQSourceConnector

Not sure on how to proceed since the error seems very generic one.

Complete error trace :

[2019-07-25 10:35:46,955] ERROR WorkerSourceTask{id=mq-source1-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
java.lang.ClassCastException: class com.ibm.eventstreams.connect.mqsource.MQSourceConnector
at java.lang.Class.asSubclass(Class.java:3404)
at com.ibm.eventstreams.connect.mqsource.JMSReader.configure(JMSReader.java:168)
at com.ibm.eventstreams.connect.mqsource.MQSourceTask.start(MQSourceTask.java:81)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:198)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
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)
[2019-07-25 10:35:46,956] ERROR WorkerSourceTask{id=mq-source1-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:178)

connect-distributed.properties file :

group.id=connect-cluster
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
rest.host.name=localhost
rest.port=8085
rest.advertised.host.name=localhost
rest.advertised.port=8085

Regards

[AndrewJSchofield]

Hi, please take the latest version of the connector, run it again and open a new issue if you see the problem. That will help me understand what's going on. Thanks, Andrew