{"record":{"id":"2e72ea36945175d9","repo":"apache/beam","slug":"attempting-to-add-message-message-to-checkpoint-that-is","errorCode":null,"errorMessage":"Attempting to add message ${message} to checkpoint that is discarded.","messagePattern":"Attempting to add message (.+?) to checkpoint that is discarded\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java","lineNumber":174,"sourceCode":"   */\n  static class Preparer {\n    private Instant oldestMessageTimestamp = Instant.now();\n    private transient List<Message> messages = new ArrayList<>();\n    private final AcknowledgeMode acknowledgeMode;\n\n    @VisibleForTesting transient boolean discarded = false;\n\n    @VisibleForTesting final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();\n\n    private Preparer(AcknowledgeMode acknowledgeMode) {\n      this.acknowledgeMode = acknowledgeMode;\n    }\n\n    void add(Message message) throws JMSException {\n      lock.writeLock().lock();\n      try {\n        if (discarded) {\n          throw new IllegalStateException(\n              String.format(\n                  \"Attempting to add message %s to checkpoint that is discarded.\", message));\n        }\n        Instant currentMessageTimestamp = new Instant(message.getJMSTimestamp());\n        if (currentMessageTimestamp.isBefore(oldestMessageTimestamp)) {\n          oldestMessageTimestamp = currentMessageTimestamp;\n        }\n        if (acknowledgeMode == AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE) {\n          messages.add(message);\n        } else {\n          // Jms spec will implicitly acknowledge _all_ messaged already received by the same\n          // session if one message in this session is being acknowledged. Only need to ack\n          // last seen one.\n          if (messages.isEmpty()) {\n            messages.add(message);\n          } else {\n            messages.set(0, message);\n          }","sourceCodeStart":156,"sourceCodeEnd":192,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java#L156-L192","documentation":"JmsCheckpointMark.add() records received messages so they can be acknowledged when the checkpoint is finalized. Once the checkpoint mark has been discarded (discarded=true, e.g. after finalize/acknowledge processing or when the unbounded source abandons it), adding further messages is a programming/state error and this IllegalStateException is thrown.","triggerScenarios":"Calling the internal add(Message) method on a JmsCheckpointMark whose discarded flag is true — i.e. after the mark was already finalized/acknowledged or discarded by the source, then more messages are added to it.","commonSituations":"Seen in Beam JMS unbounded source internals when a consumer/reader races session recreation or checkpointing after the mark was completed; also triggered by tests exercising acknowledge paths (individual/client acknowledge modes). End users hit it only via custom code touching checkpoint marks.","solutions":["Do not reuse a JmsCheckpointMark after it has been finalized or discarded; create/obtain a fresh checkpoint mark for new messages","Guard the add() call with the mark's discarded state (or synchronization) before adding messages","If seen in a pipeline, upgrade Beam — this indicates an internal race in the JMS unbounded source; file an issue with the pipeline logs","Ensure only the owning UnboundedJmsReader manages the checkpoint mark lifecycle; don't share it across readers/threads"],"exampleFix":"// before\ncheckpointMark.add(message); // may throw if already discarded\n// after\nif (!checkpointMark.isDiscarded()) { // or synchronize on the mark's lifecycle\n  checkpointMark.add(message);\n} else {\n  checkpointMark = createNewCheckpointMark();\n  checkpointMark.add(message);\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  mark.add(message);\n} catch (IllegalStateException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"checkpoint that is discarded\")) {\n    mark = createFreshCheckpointMark(); // do not reuse discarded marks\n    mark.add(message);\n  } else throw e;\n}","preventionTips":["Treat JmsCheckpointMark as single-use: never add messages after finalize/acknowledge","Let the owning UnboundedJmsReader manage mark lifecycle; don't share across threads","Keep Beam up to date — races in checkpoint discard handling are fixed upstream"],"tags":["jms","checkpoint","state","beam"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}