Migrating from JMS

JMS (Jakarta / Java Message Service) is a Java API specification, not a wire protocol. Any conforming JMS provider can be swapped for another without touching application code — your program calls Session.createProducer, Session.createConsumer, and so on regardless of the underlying transport. The migration path is therefore a ConnectionFactory swap: replace your current provider's factory with Apache Qpid JMS, which speaks AMQP 1.0 over the wire, and point it at KubeMQ's AMQP 1.0 connector. The JMS calls you already wrote stay exactly as they are.

This guide covers the connection-factory migration, the concept-to-pattern mapping, a self-contained Java snippet, the security posture, and the documented gaps. For the AMQP 1.0 wire contract behind Qpid JMS — frame-level semantics, link options, message translation, selector grammar, and settlement rules — see the Migrating from AMQP 1.0 guide and the AMQP 1.0 connector capabilities reference.

Overview

KubeMQ's AMQP 1.0 connector is the substrate. Qpid JMS connects to it over AMQP 1.0 on port 5672 (plain / SASL) or 5671 (TLS), sharing the listener with AMQP 0.9.1.

The AMQP 1.0 connector is disabled by default — a stock kubemq-server does not bind the AMQP 1.0 listener until you turn it on. Enable it with its enable variable:

docker run -d \  --name kubemq \  -p 5672:5672 \  -p 5671:5671 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_AMQP10_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Compatibility Matrix

This is the JMS column of the cross-protocol matrix in the migration hub, restated here so this guide stands on its own.

¹ Durable subscription is backed by the persistence engine. session.unsubscribe() is node-local — see What Does NOT Migrate for the full explanation.

² No client-settable DLQ — see What Does NOT Migrate for the full explanation.

Connection / Endpoint Migration

As shown in the Compatibility Matrix (Drop-in level: client-swap), the only required change is the ConnectionFactory URL and its class name. The JMS API calls (createSession, createProducer, createConsumer, and so on) remain unchanged.

Using JNDI properties

# jndi.properties — before (ActiveMQ / Artemis example)
# java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
# connectionFactory.kubemq=tcp://old-broker:61616

# jndi.properties — after (Qpid JMS over KubeMQ AMQP 1.0)
java.naming.factory.initial=org.apache.qpid.jms.jndi.JmsInitialContextFactory
connectionfactory.kubemq=amqp://kubemq.example.com:5672

# Optional TLS endpoint
# connectionfactory.kubemq=amqps://kubemq.example.com:5671

# Declare destinations (prefix drives pattern selection)
queue.ordersQueue=queues/orders
topic.ordersTopic=events/orders
topic.auditTopic=events-store/audit

Direct instantiation (no JNDI)

import org.apache.qpid.jms.JmsConnectionFactory;

// Before (example from any other provider):
// ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://old-broker:61616");

// After (Qpid JMS):
ConnectionFactory cf = new JmsConnectionFactory("amqp://kubemq.example.com:5672");

For TLS, use "amqps://kubemq.example.com:5671".

Maven / Gradle dependency

<!-- Maven — jakarta namespace (JMS 3.x) -->
<dependency>
  <groupId>org.apache.qpid</groupId>
  <artifactId>qpid-jms-client</artifactId>
  <version>2.6.0</version>
  <classifier>jakarta</classifier>
</dependency>

<!-- javax namespace (JMS 2.0), if still on javax.jms -->
<!-- <artifactId>qpid-jms-client</artifactId><version>1.12.0</version> -->

// Gradle — jakarta namespace (JMS 3.x)
implementation 'org.apache.qpid:qpid-jms-client:2.6.0:jakarta'

// javax namespace (JMS 2.0), if still on javax.jms
// implementation 'org.apache.qpid:qpid-jms-client:1.12.0'

Concept & Destination Mapping

Qpid JMS resolves destination names against KubeMQ by the address prefix — the prefix is stripped and the remainder becomes the channel name. JMS Queue / Topic objects carry the full address, not just the channel name.

Bare address (no prefix). Qpid JMS advertises a JMS capability hint (queue or topic) in the AMQP ATTACH. The connector uses this to resolve Queue("orders") → Queues and Topic("orders") → Events automatically, without requiring the prefix. If you use addresses that lack the prefix and the capability hint is absent, the connector falls back to the configured default pattern (queues). Prefer the explicit prefix so resolution is never ambiguous.

Selectors. Pass the selector expression as the second argument to createConsumer or createDurableConsumer. The connector evaluates a SQL92 subset (comparisons, AND / OR / NOT, BETWEEN, IN, LIKE, IS NULL) against application properties and JMS headers (JMSPriority, JMSType, JMSCorrelationID, JMSMessageID, JMSTimestamp). Selectors are supported on Events and Events Store links only; a selector on a Queue (queues/) link fails the ATTACH with amqp:not-implemented.

Canonical Client Example

Client: Apache Qpid JMS 2.6.0 (jakarta). API symbols: JmsConnectionFactory, Connection.createSession, Session.AUTO_ACKNOWLEDGE, Session.createQueue, Session.createTopic, Session.createProducer, Session.createConsumer, Session.createDurableConsumer, Session.createTemporaryQueue, MessageProducer.send, MessageConsumer.receive.

import jakarta.jms.*;
import org.apache.qpid.jms.JmsConnectionFactory;

public class KubeMQJmsExample {

    public static void main(String[] args) throws JMSException {

        // 1. ConnectionFactory — the only change from your previous provider
        ConnectionFactory cf = new JmsConnectionFactory(
            "amqp://kubemq.example.com:5672");

        // When authentication is enabled, pass the KubeMQ JWT as the password.
        // The username is recorded for audit only and does not affect identity.
        Connection conn = cf.createConnection("svc-orders", System.getenv("KUBEMQ_JWT"));
        conn.start();

        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // ── Point-to-point (Queues) ──────────────────────────────────────────
        Queue ordersQueue = session.createQueue("queues/orders");

        // Produce
        TextMessage msg = session.createTextMessage("order-payload");
        session.createProducer(ordersQueue).send(msg);

        // Consume (competing consumer — AUTO_ACKNOWLEDGE maps to AMQP ACCEPTED)
        TextMessage received = (TextMessage) session.createConsumer(ordersQueue).receive(5000);
        System.out.println("Received: " + received.getText());

        // ── Non-durable pub/sub (Events) ─────────────────────────────────────
        Topic eventsTopic = session.createTopic("events/notifications");
        session.createProducer(eventsTopic).send(
            session.createTextMessage("event-payload"));
        // Subscribers on events/notifications receive a copy (fan-out).

        // ── Durable subscription (Events Store) ──────────────────────────────
        Topic auditTopic = session.createTopic("events-store/audit");
        // Durable identity derived from (container-id, subscription-name) — survives reconnect.
        MessageConsumer durableConsumer =
            session.createDurableConsumer(auditTopic, "audit-sub");

        // ── Selector on Events Store ──────────────────────────────────────────
        // SQL92 subset evaluated in the connector before delivery.
        MessageConsumer filtered = session.createConsumer(
            session.createTopic("events-store/audit"),
            "JMSPriority > 4 AND region = 'EU'");

        // ── Request / reply (Commands) ────────────────────────────────────────
        // Use a TemporaryQueue as the reply destination; the connector backs it
        // with a dynamic AMQP node (/responses/<id>).
        Queue statusCmd = session.createQueue("commands/status");
        TemporaryQueue replyQ = session.createTemporaryQueue();
        Message req = session.createTextMessage("get-status");
        req.setJMSReplyTo(replyQ);
        req.setJMSCorrelationID("req-001");
        session.createProducer(statusCmd).send(req);

        Message reply = session.createConsumer(replyQ).receive(5000);
        System.out.println("Reply: " + ((TextMessage) reply).getText());

        conn.close();
    }
}

Security

Authentication

• Opt-in connector. The AMQP 1.0 connector is disabled by default. Set CONNECTORS_AMQP10_ENABLE=true (or connectors.amqp10.enable = true in TOML) to start it.

• No auth (development). When authentication is disabled (the server default), the connector accepts SASL PLAIN with any credentials and also accepts SASL ANONYMOUS. If the server is reachable from untrusted networks, enable authentication or firewall ports 5672 / 5671.

• With auth enabled. Pass a KubeMQ JWT as the SASL PLAIN password:

  Connection conn = cf.createConnection("any-username", kubemqJWT);

  The username is recorded for audit; the JWT's ClientID claim becomes the connection identity for authorization checks.

Authorization

With authorization enabled, the ClientID derived from the JWT is checked against the policy on the resolved channel at ATTACH time:

• Producers (sender links) need Write on the channel.
• Consumers (receiver links) need Read on the channel.

A denied ATTACH fails with amqp:unauthorized-access; the JMS client throws a JMSSecurityException.

TLS / mTLS

• TLS (server auth). Change the connection URL to amqps://kubemq.example.com:5671. The Qpid JMS client performs the standard TLS handshake; trust the server certificate using a JVM truststore or the transport.trustStoreLocation / transport.trustStorePassword URI options:

  amqps://kubemq.example.com:5671?transport.trustStoreLocation=/path/to/truststore.jks

• mTLS (client auth). When the server requests a client certificate, the connector offers SASL EXTERNAL; the certificate CN becomes the ClientID (no JWT needed):

  amqps://kubemq.example.com:5671?transport.keyStoreLocation=/path/to/keystore.jks&transport.keyStorePassword=secret

The TLS listener on port 5671 is active only when the server Security block is configured. See Auth & Security for the shared connector security model.

SASL mechanisms

What Does NOT Migrate / Deviations

Not supported

Behavioral deviations

What does migrate cleanly

• JMS AUTO_ACKNOWLEDGE and CLIENT_ACKNOWLEDGE modes.
• Queue (point-to-point), Topic (fan-out), and durable Topic subscriptions.
• TextMessage, BytesMessage, MapMessage, and ObjectMessage body types (round-trip lossless for standard properties).
• Selectors on Events / Events Store.
• JMSReplyTo + JMSCorrelationID for request / reply over a TemporaryQueue.
• JMSPriority, JMSTimestamp, JMSMessageID, JMSType, and JMSCorrelationID headers.

Verification Smoke Test

Use the point-to-point block from the Canonical Client Example as a send / receive confirmation:

1. Enable the AMQP 1.0 connector on the target KubeMQ server (CONNECTORS_AMQP10_ENABLE=true) and confirm port 5672 is reachable.
2. Produce one message — run the snippet's "Point-to-point (Queues)" block to publish a TextMessage to queues/orders.
3. Consume it — run the createConsumer(ordersQueue).receive(5000) block and confirm the text matches.
4. Check the dashboard — the KubeMQ dashboard AMQP 1.0 page should show one connection, one sender link, one receiver link, and one transfer in / out.
5. Confirm zero dropped messages — kubemq_amqp10_events_dropped_no_credit_total and kubemq_amqp10_events_store_dropped_stalled_total should remain 0.

See Also