apache/beam · error · IOException

Error creating JMS consumer

Error message

Error creating JMS consumer

What it means

UnboundedJmsReader.recreateSession() wraps failures from session.createConsumer(...) (on a Topic or Queue) in an IOException with the message 'Error creating JMS consumer'. The session was created but the broker refused or failed to create the MessageConsumer for the configured topic/queue.

Source

Thrown at sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:731

        int ackMode = getAckModeCode(source.spec.getAcknowledgeMode());
        this.session = this.connection.createSession(false, ackMode);
      } catch (Exception e) {
        throw new IOException("Error creating JMS session", e);
      }

      Read<T> spec = source.spec;
      Duration receiveTimeout =
          MoreObjects.firstNonNull(source.spec.getReceiveTimeout(), Duration.ZERO);
      receiveTimeoutMillis = receiveTimeout.getMillis();

      try {
        if (source.spec.getTopic() != null) {
          consumer = session.createConsumer(session.createTopic(spec.getTopic()));
        } else {
          consumer = session.createConsumer(session.createQueue(spec.getQueue()));
        }
      } catch (Exception e) {
        throw new IOException("Error creating JMS consumer", e);
      }
    }

    private int getAckModeCode(AcknowledgeMode mode) {
      if (mode == AcknowledgeMode.CLIENT_ACKNOWLEDGE
          || mode == AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE) {
        return Session.CLIENT_ACKNOWLEDGE;
      } else if (mode == AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE) {
        Integer configuredCode = source.spec.getIndividualAcknowledgeModeCode();
        if (configuredCode != null) {
          return configuredCode;
        }
        String connectionClassName = this.connection.getClass().getName();
        if (connectionClassName.contains("org.apache.activemq.ActiveMQConnection")) {
          return 4;
        } else if (connectionClassName.contains("org.apache.qpid.jms")) {
          return 101;
        } else if (connectionClassName.contains("org.apache.activemq.artemis")) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Check the wrapped cause to see whether it is destination-not-found, security, or connection related and fix accordingly
2. Verify the topic/queue name in JmsIO.read().withTopic()/withQueue() exactly matches a destination on the broker (or enable auto-creation)
3. Ensure the JMS user has consume permission on the destination
4. Check broker consumer limits and drop stale consumers if exhausted

Example fix

// before
JmsIO.<String>read().withQueue("order.Events") // destination doesn't exist / invalid name
// after
JmsIO.<String>read().withQueue("order.events") // matches actual broker destination
// (or pre-create the destination / grant consume permission)

// preflight: verify destination exists and is consumable
Destination d = session.createQueue(queueName);
try (MessageConsumer probe = session.createConsumer(d)) { /* ok */ }

try {
  reader.recreateSession();
} catch (IOException e) {
  if (e.getMessage() != null && e.getMessage().equals("Error creating JMS consumer")) {
    // inspect e.getCause(): destination missing, permissions, or limits; fix and retry
  } else throw e;
}

When it happens

Trigger: recreateSession() executing session.createConsumer(session.createTopic(topic)) or (session.createQueue(queue)) when the destination does not exist/cannot be created, the destination name is invalid, or the broker rejects the consumer (permissions, limits, connection issues).

Common situations: Typo in topic/queue name or destination not provisioned on the broker; insufficient permissions to consume from the destination; broker-side consumer limits reached; destination name containing invalid characters for the broker (e.g. SQL Server/Azure Service Bus style names on ActiveMQ).

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• Error creating JMS session
• RuntimeException
• Unable to encode constant members of ParamWindowedValueCoder
• Unable to decode constant members from payload for ParamWind
• ConnectionFactory ${className} does not exist. If using expa

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/4856aad2e50626c9. Report an issue: GitHub.