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.

Quick start: repoint one client

Try the switch on your own machine before you plan the migration.

The Docker tab points one client at the KubeMQ you started with Try KubeMQ, turning its connector on where needed and keeping your evaluation or license and your data. The Kubernetes tab works on a cluster you installed with Install on Kubernetes. Use it to try the switch, not for production; see Plans compared for staging and production licenses. If your current broker runs on this machine, stop it first, because two programs cannot publish the same port. The kmq command runs natively on Windows; the other commands are bash, so on Windows run them in WSL (Windows Subsystem for Linux).

kmq deploy update --installation kubemq --connector amqp10 --deadline 10m

{
  "installation_id": "…",
  "state": "…",
  "target": "…",
  "data_retained": true,
  "next_action": "…"
}

// Any other JMS provider, for example ActiveMQ
ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://old-broker:61616");

// Qpid JMS over KubeMQ's AMQP 1.0 connector
ConnectionFactory cf = new JmsConnectionFactory("amqp://localhost:5672");

//DEPS org.apache.qpid:qpid-jms-client:1.16.0

import javax.jms.Connection;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.qpid.jms.JmsConnectionFactory;

public class SmokeTest {
    public static void main(String[] args) throws Exception {
        // The only KubeMQ-specific line: Qpid JMS pointed at the AMQP 1.0 connector.
        JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672");
        try (Connection connection = factory.createConnection()) {
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("queues/smoke-test");
            session.createProducer(queue).send(session.createTextMessage("hello kubemq"));
            Message received = session.createConsumer(queue).receive(10_000);
            if (received == null) {
                throw new IllegalStateException("no message within 10 seconds");
            }
            System.out.println("received: " + received.getBody(String.class));
        }
    }
}

jbang SmokeTest.java

received: hello kubemq

amqp10:
  enabled: true

kmq deploy update --installation messaging --kube-context YOUR_KUBE_CONTEXT --values cluster-values.yaml --deadline 10m

{
  "installation_id": "…",
  "state": "…",
  "target": "…",
  "data_retained": true,
  "next_action": "…"
}

kubectl --context YOUR_KUBE_CONTEXT -n kubemq get svc messaging-amqp10

NAME     TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)          AGE
messaging-amqp10    ClusterIP   …            <none>        5672/TCP,…     …

kubectl --context YOUR_KUBE_CONTEXT -n kubemq port-forward svc/messaging-amqp10 5672:5672

Forwarding from 127.0.0.1:5672 -> 5672
Forwarding from [::1]:5672 -> 5672

Before you move production traffic

• Read What does not migrate before you commit.
• Plan how traffic moves and how you roll back: run both brokers, move one queue or topic at a time, and keep your JMS provider until you have verified the move.
• Install KubeMQ for real: Choose your path.

Compatibility Matrix

This is the JMS column of the cross-protocol matrix on Migrate from another broker, 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

The examples on this page use qpid-jms-client 2.6.0 (jakarta) or 1.16.0 (javax); newer releases in the same line are expected to work.

<!-- 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.16.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.16.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.x (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. Port 5672 is open regardless, because the AMQP 0-9-1 connector sharing it is on by default — so an open port does not by itself mean AMQP 1.0 is enabled.

• 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

For the basic round trip, see the quick start; this section adds deeper checks.

1. 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.
2. Confirm zero dropped messages — kubemq_amqp10_events_dropped_no_credit_total and kubemq_amqp10_events_store_dropped_stalled_total should remain 0.

See Also