Problem with the PubSub+ Micro-Integration for JMS connector to connect with JMS using activemq-client

Connectors & Integrations
,
1

Hello everyone, and thanks in advance for your help.

I’m experiencing issues using the “PubSub+ Micro-Integration for JMS” connector https://solace.com/integration-hub/pubsub-connector-for-jms/

Our use case is to consume messages from a topic via JMS from a third-party product, and send those events to Solace.

For now, to reduce setup complexity and to facilitate continuous configuration changes, I’m running the connector locally. I have no issues connecting to Solace, but the JMS read integration is not working.

Problem details we’re encountering:

The documentation indicates in several places that we must provide the ActiveMQ client libraries and dependencies:

“Download the JAR files for the JMS client and all its required dependencies and place them in the libs directory.
To use the connector with ActiveMQ, download the activemq-client and its required dependencies from Maven Central.”

Currently, the library version we’re using is:
org.apache.activemq
activemq-client
6.1.6

The issue arises because the connector pubsubplus-connector-jms-2.7.2 tries to set up the subscriber using the following code:

From the class:
pubsubplus-connector-jms-2.7.2.jar\BOOT-INF\lib\spring-cloud-stream-binder-jms-3.1.10.jar\com\solace\spring\cloud\stream\binder\jms\inbound\JmsReceiver.class
Method: refreshJMSResourcesIfNecessary()

if (this.destinationType == DestinationType.TOPIC) {
    if (this.jmsProvider == JMSProvider.ORACLEAQ) {
        logger.info("Creating durable consumer on topic [ {} ] and durableSubscription [ {} ]", this.destinationName, this.durableSubscriptionName);
        this.consumer = this.session.createDurableSubscriber(this.session.createTopic(this.destinationName), this.durableSubscriptionName);
    } else {
        logger.info("Creating shared durable consumer on topic [ {} ] and durableSubscription [ {} ]", this.destinationName, this.durableSubscriptionName);
        this.consumer = this.session.createSharedDurableConsumer(this.session.createTopic(this.destinationName), this.durableSubscriptionName);
    }
} else {
    logger.info("Creating consumer on queue [ {} ]", this.destinationName);
    this.consumer = this.session.createConsumer(this.session.createQueue(this.destinationName));
}

Since this is a JMS topic and not ORACLEAQ, the connector executes:
this.session.createSharedDurableConsumer(this.session.createTopic(this.destinationName), this.durableSubscriptionName);

Here, Session is the jakarta.jms.Session interface.

The options available in the activemq-client library are:

  • org.apache.activemq.ActiveMQSession is the one being used, and its implementation throws:
    throw new UnsupportedOperationException("createSharedDurableConsumer(Topic, name) is not supported");
  • org.apache.activemq.ActiveMQQueueSession is not being used and also does not implement it.
  • org.apache.activemq.ActiveMQTopicSession is also not being used and does not implement it either.

We are aware of other libraries that implement this functionality, such as artemis-jms-client, but that one implements the javax.jms.Session interface, and it’s not aligned with what the Solace documentation recommends.

We’re running the connector with:

java -Dloader.path=libs/ -Dspring.config.additional-location=file:./samples/config/ -jar pubsubplus-connector-jms-2.7.2.jar

lib folder contents:

activemq-client-6.1.6.jar  
hawtbuf-1.11.jar  
jakarta.jms-api-3.1.0.jar  
jmdns-3.6.0.jar  
slf4j-api-2.0.16.jar  
xstream-1.4.21.jar  
README.md  
README.pdf

ERROR:
com.solace.spring.cloud.stream.binder.jms.inbound.JmsReceiver - Could not refresh consumer JMS resources - retrying using ExponentialBackOffExecution{currentInterval=22780ms, multiplier=1.5, attempts=7}. Cause: createSharedDurableConsumer(Topic, name) is not supported

Questions:

  • Is it not possible to use ActiveMQ with a topic in this context?
  • What alternative JMS libraries can be used for topic consumption that are compatible with the connector?