{"record":{"id":"2ad10b295616298e","repo":"apache/iceberg","slug":"unexpected-command-type-in-keyed-stream-2ad10b","errorCode":null,"errorMessage":"Unexpected command type in keyed stream: ","messagePattern":"Unexpected command type in keyed stream: ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java","lineNumber":192,"sourceCode":"        ctx.timerService().registerEventTimeTimer(ts);\n      } else if (cmd.type() == IndexCommand.Type.RESOLVE_DELETE) {\n        Long resolveTs = resolveTimestamp.value();\n        if (resolveTs == null || ts > resolveTs) {\n          resolveTimestamp.update(ts);\n        }\n\n        Long currentSeq = resolveSequenceNumber.value();\n        if (currentSeq == null || cmd.deleteSequenceNumber() > currentSeq) {\n          resolveSequenceNumber.update(cmd.deleteSequenceNumber());\n        }\n\n        // Accumulate every delete's scope.\n        // One delete phase can carry same-key deletes from multiple specs.\n        resolveSpecIds.add(cmd.deleteSpecId());\n\n        ctx.timerService().registerEventTimeTimer(ts);\n      } else {\n        throw new IllegalStateException(\"Unexpected command type in keyed stream: \" + cmd.type());\n      }\n    } catch (Exception e) {\n      LOG.error(\"PKIndex failed to process command of type {}\", cmd.type(), e);\n      ctx.output(TaskResultAggregator.ERROR_STREAM, e);\n      out.collect(DVPosition.ABORT);\n    }\n  }\n\n  @Override\n  public void processBroadcastElement(IndexCommand cmd, Context ctx, Collector<DVPosition> out) {\n    Preconditions.checkArgument(\n        cmd.type() == IndexCommand.Type.CLEAR_INDEX,\n        \"Broadcast element must be %s\",\n        IndexCommand.Type.CLEAR_INDEX);\n\n    final long broadcastGeneration = cmd.indexGeneration();\n    try {\n      ctx.applyToKeyedState(","sourceCodeStart":174,"sourceCodeEnd":210,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java#L174-L210","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure all Flink operators of the maintenance job run the same Iceberg version; restart the job from a fresh (or same-version) savepoint.","Check the logged command type (LOG.error prints cmd.type()) and confirm the planner does not emit it into the keyed stream.","If you customized the planner to add a command type, add the corresponding handling branch in EqualityConvertPKIndex.processElement.","Rebuild the topology so commands are routed only to the operators that understand them."],"exampleFix":"// before: mixed versions after savepoint restore\nflink run -s old-savepoint.jar iceberg-flink-1.6-job.jar\n// after\nflink run -s old-savepoint.jar iceberg-flink-1.9-job.jar // same version as savepoint","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"// before routing a command into the keyed stream\nif (cmd.type() != CMD_DELETE && cmd.type() != CMD_EXPECTED) {\n  throw new IllegalArgumentException(\"PKIndex cannot handle command type \" + cmd.type());\n}","tryCatchPattern":"// PKIndex already catches Exception internally and emits ABORT on the error stream\nSingleOutputStreamOperator<?> result = keyed.process(new EqualityConvertPKIndex(...))\n    .getSideOutput(TaskResultAggregator.ERROR_STREAM)\n    .filter(e -> e instanceof IllegalStateException) // handle/abort job\n    .startNewChain();","preventionTips":["Run all maintenance operators from the same Iceberg build","Don't restore savepoints across different library versions without validating the job graph","If adding new command types in the planner, update every consumer (PKIndex, Reader)","Keep command routing centralized to avoid misrouted events"],"tags":["flink","internal-protocol","state-machine"],"backgroundTag":"invalid-enum-value","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}