-
Notifications
You must be signed in to change notification settings - Fork 85
Open
Labels
Description
What happened?
We try to use the JMSCorrelationID as Kafka record key, because we need it for our answer. But as soon as JMSCorrelationID
as mq.record.builder.key.header
, the connector is crashing with the following stacktace:
java.lang.NullPointerException
at com.ibm.eventstreams.connect.mqsource.builders.BaseRecordBuilder.getKey(BaseRecordBuilder.java:115)
at com.ibm.eventstreams.connect.mqsource.builders.BaseRecordBuilder.toSourceRecord(BaseRecordBuilder.java:166)
at com.ibm.eventstreams.connect.mqsource.JMSReader.receive(JMSReader.java:282)
at com.ibm.eventstreams.connect.mqsource.MQSourceTask.poll(MQSourceTask.java:126)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.poll(AbstractWorkerSourceTask.java:462)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:351)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75)
at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
CorrelationID is set in the message. All others option work fine, including JMSCorrelationIDAsBytes
.
Steps To Reproduce
- Setup connector with settings
mq.message.body.jms: true
mq.record.builder.key.header: JMSCorrelationID
key.converter: org.apache.kafka.connect.storage.StringConverter
com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder - Put message on queue in format MQSTR with Correlation- and MessageID set
Environment
- v1.3.4 of kafka-connect-mq-source
Anything else?
No response
Version
1.3.4 (Default)
Relevant log output
java.lang.NullPointerException
at com.ibm.eventstreams.connect.mqsource.builders.BaseRecordBuilder.getKey(BaseRecordBuilder.java:115)
at com.ibm.eventstreams.connect.mqsource.builders.BaseRecordBuilder.toSourceRecord(BaseRecordBuilder.java:166)
at com.ibm.eventstreams.connect.mqsource.JMSReader.receive(JMSReader.java:282)
at com.ibm.eventstreams.connect.mqsource.MQSourceTask.poll(MQSourceTask.java:126)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.poll(AbstractWorkerSourceTask.java:462)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:351)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75)
at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
Code of Conduct
- I agree to follow this project's Code of Conduct