apache/iceberg · error · IllegalStateException

Unexpected command type in keyed stream: <cmd.type()>

Error message

Unexpected command type in keyed stream: <cmd.type()>

What it means

The PK-index keyed operator received a command whose type is neither a data-file read nor a delete to accumulate, so the state machine cannot advance. This is an internal invariant violation: upstream planner commands and the keyed operator's understood command set have diverged. The operator logs the error, emits it on the error stream, and collects DVPosition.ABORT to unwind the cycle.

Source

Thrown at flink/v1.20/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. Verify all Flink maintenance modules (planner, PK index, reader, writer) come from the same Iceberg version - do not mix jars across versions.
2. Inspect TaskResultAggregator.ERROR_STREAM output to see the logged command type and find which operator emitted it.
3. Upgrade/align the whole connector to a version where planner and PKIndex command handling match.
4. If you extended commands locally, add the corresponding case in EqualityConvertPKIndex.processElement.

Example fix

// before: mixed jars: iceberg-flink 1.20 planner + patched PKIndex handling only DATA/DELETE
// after: use one consistent iceberg-flink version, or handle the new type:
} else if (cmd.type() == CommandType.RESOLVE_DELETE) {
  handleResolveDelete(cmd, ctx);
}

// before emitting commands, ensure the deployed version set is uniform
String v = EqualityConvertPKIndex.class.getPackage().getImplementationVersion();
if (!Objects.equals(v, plannerVersion)) throw new IllegalStateException("Mixed connector versions: " + v + " vs " + plannerVersion);

boolean isSupportedCommand(ReadCommand cmd) {
  return cmd != null && (cmd.type() == CommandType.DATA || cmd.type() == CommandType.DELETE);
}

// operator already routes to error stream; on consumer side
DataStream<Throwable> errors = result.getErrorStream();
errors.process((ctx, t) -> {
  if (msg.contains("Unexpected command type")) alertMixedVersions();
});

When it happens

Trigger: A ReadCommand with an unexpected/unknown type (e.g. a newly added command type not handled in EqualityConvertPKIndex.processElement) is emitted by the planner into the keyed stream.

Common situations: Mixed Flink/Iceberg versions where the planner emits new command types the operator doesn't know; a bug in the planner emitting an uninitialized command; custom modifications to the command enum.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• Unexpected ContentScanTask type: <task.getClass().getName()>
• Unexpected command type in keyed stream:
• Internal algorithm error: exhausted subtasks with unassigned
• Unexpected command type in keyed stream:
• Illegal table name:

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