apache/beam · error · IllegalStateException

Attempting to add message

Error message

Attempting to add message ${message} to checkpoint that is discarded.

What it means

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.

Solutions

1. Do not reuse a JmsCheckpointMark after it has been finalized or discarded; create/obtain a fresh checkpoint mark for new messages
2. Guard the add() call with the mark's discarded state (or synchronization) before adding messages
3. If seen in a pipeline, upgrade Beam — this indicates an internal race in the JMS unbounded source; file an issue with the pipeline logs
4. Ensure only the owning UnboundedJmsReader manages the checkpoint mark lifecycle; don't share it across readers/threads

Example fix

// before
checkpointMark.add(message); // may throw if already discarded
// after
if (!checkpointMark.isDiscarded()) { // or synchronize on the mark's lifecycle
  checkpointMark.add(message);
} else {
  checkpointMark = createNewCheckpointMark();
  checkpointMark.add(message);
}

try {
  mark.add(message);
} catch (IllegalStateException e) {
  if (e.getMessage() != null && e.getMessage().contains("checkpoint that is discarded")) {
    mark = createFreshCheckpointMark(); // do not reuse discarded marks
    mark.add(message);
  } else throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Cannot access state in non-window observing context.
• ConnectionFactory does not exist. If using expansion…
• Error creating JMS consumer
• Error creating JMS session
• error decoding append bag user state window key

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

Appendix: source

Thrown at sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:174

   */
  static class Preparer {
    private Instant oldestMessageTimestamp = Instant.now();
    private transient List<Message> messages = new ArrayList<>();
    private final AcknowledgeMode acknowledgeMode;

    @VisibleForTesting transient boolean discarded = false;

    @VisibleForTesting final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    private Preparer(AcknowledgeMode acknowledgeMode) {
      this.acknowledgeMode = acknowledgeMode;
    }

    void add(Message message) throws JMSException {
      lock.writeLock().lock();
      try {
        if (discarded) {
          throw new IllegalStateException(
              String.format(
                  "Attempting to add message %s to checkpoint that is discarded.", message));
        }
        Instant currentMessageTimestamp = new Instant(message.getJMSTimestamp());
        if (currentMessageTimestamp.isBefore(oldestMessageTimestamp)) {
          oldestMessageTimestamp = currentMessageTimestamp;
        }
        if (acknowledgeMode == AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE) {
          messages.add(message);
        } else {
          // Jms spec will implicitly acknowledge _all_ messaged already received by the same
          // session if one message in this session is being acknowledged. Only need to ack
          // last seen one.
          if (messages.isEmpty()) {
            messages.add(message);
          } else {
            messages.set(0, message);
          }

View on GitHub (pinned to 12126d8942)