{"record":{"id":"842f80a334be0524","repo":"apache/iceberg","slug":"unexpected-command-type-in-keyed-stream","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.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java#L174-L210","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check the wrapped cmd.type() in the TaskResultAggregator error output to identify the offending type","Restart the job from a fresh, consistent state (cancel with savepoint discard and rerun the cycle)","Ensure all operators run the same Iceberg/Flink version after upgrade (no mixed old/new classpath)","Report as a bug if reproducible with a single writer — the command stream contract is violated internally"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// validate command types before feeding the keyed stream\nif (cmd.type() != CMD_ADD && cmd.type() != CMD_DELETE) {\n  throw new IllegalArgumentException(\"Unsupported command type: \" + cmd.type());\n}","typeGuard":null,"tryCatchPattern":"// consumer side already funnels errors to the error stream\nresultAggregator.errorStream().subscribe(err -> {\n  if (err.getMessage().startsWith(\"Unexpected command type in keyed stream\")) {\n    // abort cycle, restart from a consistent checkpoint\n  }\n});","preventionTips":["Run uniform operator versions across the job (no mixed jars after upgrades)","Don't restore from checkpoints taken by incompatible job versions","Add unit tests covering every command type in the keyed stream"],"tags":["internal","stream","state-machine","flink"],"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"}