{"record":{"id":"d2bd83f1b2033228","repo":"apache/beam","slug":"error-creating-jms-session","errorCode":null,"errorMessage":"Error creating JMS session","messagePattern":"Error creating JMS session","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java","lineNumber":716,"sourceCode":"    // Acknowlging messages need open consumer. Tracking active checkpoints allows delayed close of\n    // session and consumer.\n    private final AtomicInteger activeCheckpoints = new AtomicInteger(0);\n\n    public UnboundedJmsReader(UnboundedJmsSource<T> source, PipelineOptions options) {\n      this.source = source;\n      this.checkpointMarkPreparer = JmsCheckpointMark.newPreparer(source.spec.getAcknowledgeMode());\n      this.currentMessage = null;\n      this.currentID = EMPTY;\n      this.options = options;\n    }\n\n    /** recreate session and consumer. */\n    private synchronized void recreateSession() throws IOException {\n      try {\n        int ackMode = getAckModeCode(source.spec.getAcknowledgeMode());\n        this.session = this.connection.createSession(false, ackMode);\n      } catch (Exception e) {\n        throw new IOException(\"Error creating JMS session\", e);\n      }\n\n      Read<T> spec = source.spec;\n      Duration receiveTimeout =\n          MoreObjects.firstNonNull(source.spec.getReceiveTimeout(), Duration.ZERO);\n      receiveTimeoutMillis = receiveTimeout.getMillis();\n\n      try {\n        if (source.spec.getTopic() != null) {\n          consumer = session.createConsumer(session.createTopic(spec.getTopic()));\n        } else {\n          consumer = session.createConsumer(session.createQueue(spec.getQueue()));\n        }\n      } catch (Exception e) {\n        throw new IOException(\"Error creating JMS consumer\", e);\n      }\n    }\n","sourceCodeStart":698,"sourceCodeEnd":734,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java#L698-L734","documentation":"UnboundedJmsReader.recreateSession() wraps any exception from connection.createSession(false, ackMode) in an IOException with the message 'Error creating JMS session'. The reader recreates the JMS session (e.g. after checkpoint finalize in CLIENT_ACKNOWLEDGE mode) and any failure talking to or authenticating with the broker surfaces as this IOException.","triggerScenarios":"reader.recreateSession() invoked while the JMS Connection is closed/broken, credentials are invalid, the broker rejects session creation (e.g. limits, shutdown), or the acknowledge mode mapping produces an unsupported session mode.","commonSituations":"Broker restarted or network drop between connection creation and session recreation; session quota exhausted on the broker; connection closed after an earlier failure and the reader attempts recreation; IBM MQ/Qpid rejecting session options.","solutions":["Verify the broker is reachable and the JMS connection is still open when the reader recreates the session (check logs just before this error)","Inspect the wrapped cause exception for the root reason (auth failure, connection closed, quota) and fix that (credentials, network, broker limits)","Make the pipeline resilient to transient broker outages by restarting the worker or using Beam's retryable failure handling","Confirm ackMode is a supported AcknowledgeMode and the broker client supports the mapped session acknowledgment constant"],"exampleFix":"// no code change in Beam; ensure broker health before/at runtime\n// before: connection broken -> createSession throws\n// after: validate/reconnect the Connection before recreation\nif (connection == null || isClosed(connection)) {\n  connection = connectionFactory.createConnection(user, password);\n  connection.start();\n}\nsession = connection.createSession(false, ackMode);","handlingStrategy":"retry","validationCode":"// preflight: ensure broker connectivity before pipeline start\ntry (Connection c = connectionFactory.createConnection(user, pass)) { c.start(); }","typeGuard":null,"tryCatchPattern":"try {\n  reader.recreateSession();\n} catch (IOException e) {\n  if (e.getMessage() != null && e.getMessage().equals(\"Error creating JMS session\")) {\n    // inspect e.getCause(); reconnect/restart with backoff for transient broker outages\n  } else throw e;\n}","preventionTips":["Monitor broker health/quotas (session limits) before and during pipeline runs","Validate credentials and network reachability to the broker from worker nodes","Choose supported AcknowledgeMode values understood by your broker client"],"tags":["jms","session","ioexception","beam","broker"],"backgroundTag":"connection-refused","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}