apache/iceberg · error · IllegalStateException

Unexpected command type in keyed stream:

Error message

Unexpected command type in keyed stream: 

What it means

EqualityConvertPKIndex.processElement() handles a keyed stream of delete-conversion commands; if a command of an unexpected type arrives in the else branch, it throws IllegalStateException. The catch block logs, emits the exception to the TaskResultAggregator error stream, and collects DVPosition.ABORT so the whole conversion task aborts cleanly.

Source

Thrown at flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java:192

        ctx.timerService().registerEventTimeTimer(ts);
      } else if (cmd.type() == IndexCommand.Type.RESOLVE_DELETE) {
        Long resolveTs = resolveTimestamp.value();
        if (resolveTs == null || ts > resolveTs) {
          resolveTimestamp.update(ts);
        }

        Long currentSeq = resolveSequenceNumber.value();
        if (currentSeq == null || cmd.deleteSequenceNumber() > currentSeq) {
          resolveSequenceNumber.update(cmd.deleteSequenceNumber());
        }

        // Accumulate every delete's scope.
        // One delete phase can carry same-key deletes from multiple specs.
        resolveSpecIds.add(cmd.deleteSpecId());

        ctx.timerService().registerEventTimeTimer(ts);
      } else {
        throw new IllegalStateException("Unexpected command type in keyed stream: " + cmd.type());
      }
    } catch (Exception e) {
      LOG.error("PKIndex failed to process command of type {}", cmd.type(), e);
      ctx.output(TaskResultAggregator.ERROR_STREAM, e);
      out.collect(DVPosition.ABORT);
    }
  }

  @Override
  public void processBroadcastElement(IndexCommand cmd, Context ctx, Collector<DVPosition> out) {
    Preconditions.checkArgument(
        cmd.type() == IndexCommand.Type.CLEAR_INDEX,
        "Broadcast element must be %s",
        IndexCommand.Type.CLEAR_INDEX);

    final long broadcastGeneration = cmd.indexGeneration();
    try {
      ctx.applyToKeyedState(

View on GitHub (pinned to 86d9c8fc54)

Solutions

1. Check the wrapped cmd.type() in the TaskResultAggregator error output to identify the offending type
2. Restart the job from a fresh, consistent state (cancel with savepoint discard and rerun the cycle)
3. Ensure all operators run the same Iceberg/Flink version after upgrade (no mixed old/new classpath)
4. Report as a bug if reproducible with a single writer — the command stream contract is violated internally

// validate command types before feeding the keyed stream
if (cmd.type() != CMD_ADD && cmd.type() != CMD_DELETE) {
  throw new IllegalArgumentException("Unsupported command type: " + cmd.type());
}

// consumer side already funnels errors to the error stream
resultAggregator.errorStream().subscribe(err -> {
  if (err.getMessage().startsWith("Unexpected command type in keyed stream")) {
    // abort cycle, restart from a consistent checkpoint
  }
});

When it happens

Trigger: A command whose type is neither ADD nor the handled delete type is fed into the keyed PK-index stream — an internal contract violation between the planner/emitter and the operator, or a corrupted/deserialized command record.

Common situations: Version-skewed job upgrade where an old operator emits command types the new operator doesn't know; checkpoint/restore replaying records from an incompatible snapshot; internal bug emitting raw commands into the wrong stream.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Unknown type for int field. Type name: ${className}
• Unknown type for binary field. Type name: ${className}
• Unexpected command type in keyed stream: <cmd.type()>
• Unknown comparison result
• Unknown comparison result

AI-assisted analysis of apache/iceberg@86d9c8fc54 (2026-09-12). Data as JSON: /api/errors/842f80a334be0524. Report an issue: GitHub.