SJMS2 connector fails with qpid JmsConnectionFactory

[fvaleri]

The SJMS2 connector works fine with org.apache.activemq.ActiveMQConnectionFactory (OpenWire protocol) and org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory (CORE protocol), but not with org.apache.qpid.jms.JmsConnectionFactory (AMQP protocol).

Reproducer:

{
    "name": "json-jms-sink",
    "config": {
        "tasks.max": "1",
        "connector.class": "org.apache.camel.kafkaconnector.sjms2.CamelSjms2SinkConnector",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter",
        "topics": "localhost.myschema.mytable",
        "camel.sink.url": "sjms2://queue:TestQueue",
        "camel.component.sjms2.connection-factory": "#class:org.apache.qpid.jms.JmsConnectionFactory",
        "camel.component.sjms2.connection-factory.brokerURL": "amqp://localhost:5672",
        "camel.component.sjms2.connection-username": "admin",
        "camel.component.sjms2.connection-password": "admin"
    }
}

Task exception:

org.apache.kafka.connect.errors.ConnectException: Failed to create and start Camel context
	at org.apache.camel.kafkaconnector.CamelSinkTask.start(CamelSinkTask.java:111)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:302)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:193)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
	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.camel.PropertyBindingException: Error binding property (camel.component.sjms2.brokerURL=amqp://localhost:5672) with name: brokerURL on bean: org.apache.qpid.jms.JmsConnectionFactory@2be09494 with value: amqp://localhost:5672
	at org.apache.camel.main.MainHelper.setPropertiesOnTarget(MainHelper.java:192)
	at org.apache.camel.main.BaseMainSupport.autoConfigurationFromProperties(BaseMainSupport.java:1282)
	at org.apache.camel.main.BaseMainSupport.autoconfigure(BaseMainSupport.java:450)
	at org.apache.camel.main.BaseMainSupport.postProcessCamelContext(BaseMainSupport.java:522)
	at org.apache.camel.main.SimpleMain.doInit(SimpleMain.java:32)
	at org.apache.camel.support.service.BaseService.init(BaseService.java:83)
	at org.apache.camel.support.service.BaseService.start(BaseService.java:111)
	at org.apache.camel.kafkaconnector.CamelSinkTask.start(CamelSinkTask.java:104)
	... 9 more```

[oscerd]

For amqp you need to use the amqp connector

Here is the sink example https://github.com/apache/camel-kafka-connector-examples/blob/master/amqp/config/CamelAmqpSinkConnector.properties

[fvaleri]

Thanks @oscerd , I added this to remember to look into it. I would expect the qpid JMS library to work, but there is probably some issue with the Camel SJMS2 component.

[valdar]

@fvaleri felle free to reopen this or another issue if you look into this further. Closing this one.