{"record":{"id":"bb5c3921e31d2a5c","repo":"apache/beam","slug":"there-are-different-initial-sequences-detected-s-and-s","errorCode":null,"errorMessage":"There are different initial sequences detected: %s and %s","messagePattern":"There are different initial sequences detected: (.+?) and (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java","lineNumber":61,"sourceCode":"    return a.isAfter(b) ? a : b;\n  }\n\n  /**\n   * The tree contains a set of non-overlapping contiguous ranges, where the key is the lower\n   * inclusive start of the range, left value of the pair is the inclusive end of the range and the\n   * right value of the pair is the maximum timestamp in the range.\n   *\n   * <p>The maximum timestamp is critical for the correctness of the ordered processing. During the\n   * merge process the merged range is assigned the maximum timestamp of the two ranges that created\n   * this new range.\n   */\n  private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();\n\n  private @Nullable Long initialSequence = null;\n\n  public void add(long sequence, Instant timestamp, boolean isInitialSequence) {\n    if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {\n      throw new IllegalStateException(\n          \"There are different initial sequences detected: \"\n              + initialSequence\n              + \" and \"\n              + sequence);\n    }\n\n    if (sequence == Long.MAX_VALUE) {\n      // This is an invalid value and DoFns will not process this element. This will also allow\n      // to produce a ContiguousSequenceRange with the exclusive end value.\n      return;\n    }\n\n    if (isInitialSequence) {\n      this.initialSequence = sequence;\n      clearRangesBelowInitialSequence(sequence, timestamp);\n    } else if (initialSequence != null && sequence <= initialSequence) {\n      // No need to add anything lower than the initial sequence to the accumulator.\n      return;","sourceCodeStart":43,"sourceCodeEnd":79,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java#L43-L79","documentation":"SequenceRangeAccumulator.add() records which sequence was the first (initial) event of a key. If an element marked as an initial sequence is added while a different initialSequence is already stored, the accumulator's invariant is violated and it throws IllegalStateException. This guards against corrupted or merged-by-mistake combine state where events of different keys/sessions share one accumulator.","triggerScenarios":"Calling add(sequence, timestamp, isInitialSequence=true) on an accumulator whose initialSequence is already non-null and different from the new sequence; typically caused by mis-partitioned input feeding one accumulator events from multiple keys, or combining accumulators incorrectly in a custom CombineFn.","commonSituations":"Custom fanout or key-breaking in the Beam combine where accumulators are reused across keys; a bug in a custom mergeAccumulators/addInput; deterministic unit tests deliberately feeding conflicting initial sequences.","solutions":["Ensure events are keyed correctly so all elements reaching one accumulator belong to the same initial sequence.","If reusing accumulators, create a fresh SequenceRangeAccumulator per key instead of reusing instances.","Fix custom addInput/mergeAccumulators logic to route by key before adding.","Verify the isInitialSequence flag is only set for the true first event of the stream."],"exampleFix":"// before\nacc.add(42, ts, true); // acc.initialSequence == 7 -> IllegalStateException\n// after\nSequenceRangeAccumulator acc = SequenceRangeAccumulator.create(); // fresh accumulator for this key\nacc.add(42, ts, true);","handlingStrategy":"validation","validationCode":"// before adding\nif (acc.hasInitialSequence() && acc.getInitialSequence() != sequence && isInitial) {\n  throw new IllegalArgumentException(\"mismatched initial sequence \" + sequence + \" vs \" + acc.getInitialSequence());\n}","typeGuard":"static boolean canAccept(SequenceRangeAccumulator acc, long sequence, boolean isInitial) {\n  return !isInitial || acc.getInitialSequence() == null || acc.getInitialSequence() == sequence;\n}","tryCatchPattern":"try {\n  acc.add(seq, ts, isInitial);\n} catch (IllegalStateException e) {\n  if (e.getMessage().startsWith(\"There are different initial sequences\")) {\n    throw new IllegalStateException(\"events of different keys reached one accumulator; fix keying\", e);\n  }\n  throw e;\n}","preventionTips":["Key events correctly so each accumulator sees one stream","Never reuse accumulator instances across keys","Set isInitialSequence=true only for the true first event"],"tags":["apache-beam","combiner","state-corruption","invariant"],"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"}