apache/iceberg · error · IllegalStateException

Unexpected command type in keyed stream:

Error message

Unexpected command type in keyed stream: 

What it means

Thrown by EqualityConvertPKIndex.processElement when a command arriving on the keyed stream is neither an expected command type (e.g. a delete command with a known type) nor one the PK index understands. It signals an internal protocol violation: the command enum value is not handled, followed by an ABORT sentinel on the output stream.

Source

Thrown at flink/v2.3/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. Ensure all Flink operators of the maintenance job run the same Iceberg version; restart the job from a fresh (or same-version) savepoint.
2. Check the logged command type (LOG.error prints cmd.type()) and confirm the planner does not emit it into the keyed stream.
3. If you customized the planner to add a command type, add the corresponding handling branch in EqualityConvertPKIndex.processElement.
4. Rebuild the topology so commands are routed only to the operators that understand them.

Example fix

// before: mixed versions after savepoint restore
flink run -s old-savepoint.jar iceberg-flink-1.6-job.jar
// after
flink run -s old-savepoint.jar iceberg-flink-1.9-job.jar // same version as savepoint

// before routing a command into the keyed stream
if (cmd.type() != CMD_DELETE && cmd.type() != CMD_EXPECTED) {
  throw new IllegalArgumentException("PKIndex cannot handle command type " + cmd.type());
}

// PKIndex already catches Exception internally and emits ABORT on the error stream
SingleOutputStreamOperator<?> result = keyed.process(new EqualityConvertPKIndex(...))
    .getSideOutput(TaskResultAggregator.ERROR_STREAM)
    .filter(e -> e instanceof IllegalStateException) // handle/abort job
    .startNewChain();

When it happens

Trigger: A ReadCommand/Command with an unrecognized type() is routed into the keyed PK-index stream, e.g. after adding a new command type to the planner without extending the PK index switch, or a corrupted/misrouted event due to pipeline wiring changes.

Common situations: Upgrading the Iceberg Flink maintenance library on some operators but not others (mixed versions in the same job after a savepoint restore); custom modifications to the maintenance pipeline that emit new command types; checkpoint/savepoint recovery from an incompatible job graph version.

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

• Unexpected command type in keyed stream: <cmd.type()>
• Unexpected command type in keyed stream:
• Unexpected ContentScanTask type:
• Illegal table name:
• Namespaces are not supported by catalog:

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