{"record":{"id":"e48330a173bec74e","repo":"apache/beam","slug":"pool-s-unknown-case-which-is-likely-a-bug-state-s-seqnum-s","errorCode":null,"errorMessage":"Pool %s - unknown case which is likely a bug: state=%s seqnum=%s","messagePattern":"Pool (.+?) - unknown case which is likely a bug: state=(.+?) seqnum=(.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/EFOShardSubscriber.java","lineNumber":183,"sourceCode":"          String lastContinuationSequenceNumber = eventsSubscriber.sequenceNumber;\n\n          // happy-path re-subscribe, subscription was complete by the SDK after 5 min\n          if (error == null && state != STOPPED && lastContinuationSequenceNumber != null) {\n            internalReSubscribe(lastContinuationSequenceNumber);\n            return;\n          }\n\n          // shard is fully consumed - re-shard happened\n          if (error == null && state != STOPPED && lastContinuationSequenceNumber == null) {\n            done.complete(null);\n            return;\n          }\n\n          String msg =\n              String.format(\n                  \"Pool %s - unknown case which is likely a bug: state=%s seqnum=%s\",\n                  pool.getPoolId(), state, lastContinuationSequenceNumber);\n          LOG.warn(\"{}\", msg);\n          done.completeExceptionally(new IllegalStateException(msg));\n        };\n  }\n\n  /**\n   * Subscribes to shard {@link #shardId} at starting position and automatically re-subscribes when\n   * necessary using {@link #reSubscriptionHandler}.\n   *\n   * <p>Note:\n   * <li>{@link #subscribe} may only ever be invoked once by an external caller.\n   * <li>The re-subscription is hidden from the external caller. To the outside it looks this\n   *     subscriber is always subscribed to the shard once {@link #subscribe} was called.\n   *\n   * @return {@link #done} to signal completion of this subscriber, normally (stopped or shard is\n   *     completely consumed) or exceptionally due to a non retry-able error.\n   */\n  CompletableFuture<Void> subscribe(StartingPosition position) {\n    checkState(state == INITIALIZED, \"Subscriber was already started\");","sourceCodeStart":165,"sourceCodeEnd":201,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/EFOShardSubscriber.java#L165-L201","documentation":"During EFO (Enhanced Fan-Out) shard subscription, the subscriber's internal state machine reached an unexpected combination of subscription state and continuation sequence number. Rather than continuing with wrong assumptions about shard position, the subscriber completes its future exceptionally with an IllegalStateException, halting consumption for that shard.","triggerScenarios":"The internal state machine in EFOShardSubscriber has cases for states like INITIALIZING/RECEIVED with-or-without a continuation sequence number; a combination falls through to the default case in the switch. This is a library-internal invariant violation, typically surfaced after resubscribing a shard or when checkpoint/sequence number handling races with state transitions.","commonSituations":"Seen when a Kinesis checkpoint contains a sequence number but the subscriber state says it should not have one (or vice versa), often after restarting from a checkpoint, shard splits/merges, or upgrades between Beam versions that changed the state machine.","solutions":["Upgrade to the latest Beam release for aws2 kinesis EFO, as this path marks a likely bug fixed in newer versions","Report/reproduce with the pool id, state and seqnum logged in the warning; check your checkpoint store for malformed or half-written sequence numbers","As a workaround, reset the shard's checkpoint so the subscriber starts from a clean position (e.g. TRIM_HORIZON or LATEST)","Verify you are not sharing/mutating the checkpoint mark across threads while EFO subscription is active"],"exampleFix":"// before: checkpoint restored with a seqnum but state machine expects none\n// after: discard stale checkpoint for the shard so subscription restarts cleanly\n// CopyOnExtendedInMemoryStateInternals: clear shard checkpoint or use InitialPositionInStream.LATEST\nconfig.setInitialPosition(InitialPositionInStream.LATEST);","handlingStrategy":"try-catch","validationCode":"// Verify checkpoint sequence numbers are well-formed before restart\nfor (ShardCheckpoint cp : checkpoints) {\n  if (cp.getSequenceNumber() != null && !ShardCheckpoint.isSentinel(cp.getSequenceNumber())) {\n    Preconditions.checkState(!cp.getSequenceNumber().isEmpty(), \"empty seqnum for %s\", cp.getShardId());\n  }\n}","typeGuard":"boolean hasValidContinuation(String state, String seqnum) {\n  return (\"RECEIVED\".equals(state)) == (seqnum != null); // state and seqnum must agree\n}","tryCatchPattern":"try {\n  subscriber.start().get();\n} catch (ExecutionException e) {\n  if (e.getCause() instanceof IllegalStateException) {\n    // reset shard checkpoint and resubscribe from a sentinel position\n    checkpointStore.reset(shardId);\n  } else { throw e; }\n}","preventionTips":["Keep Beam aws2 kinesis dependencies up to date — this path flags an internal bug","Never mutate checkpoint marks concurrently with active EFO subscriptions","Log and preserve pool id/state/seqnum output when filing bug reports","Restart from sentinel positions (LATEST/TRIM_HORIZON) after abnormal shutdowns"],"tags":["kinesis","enhanced-fan-out","state-machine","checkpoint"],"backgroundTag":"internal-invariant-violation","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"}