{"record":{"id":"6878c96439294b88","repo":"apache/iceberg","slug":"unexpected-command-type-in-keyed-stream-cmd-type","errorCode":null,"errorMessage":"Unexpected command type in keyed stream: <cmd.type()>","messagePattern":"Unexpected command type in keyed stream: <cmd\\.type\\(\\)>","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java#L174-L210","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify all Flink maintenance modules (planner, PK index, reader, writer) come from the same Iceberg version - do not mix jars across versions.","Inspect TaskResultAggregator.ERROR_STREAM output to see the logged command type and find which operator emitted it.","Upgrade/align the whole connector to a version where planner and PKIndex command handling match.","If you extended commands locally, add the corresponding case in EqualityConvertPKIndex.processElement."],"exampleFix":"// before: mixed jars: iceberg-flink 1.20 planner + patched PKIndex handling only DATA/DELETE\n// after: use one consistent iceberg-flink version, or handle the new type:\n} else if (cmd.type() == CommandType.RESOLVE_DELETE) {\n  handleResolveDelete(cmd, ctx);\n}","handlingStrategy":"validation","validationCode":"// before emitting commands, ensure the deployed version set is uniform\nString v = EqualityConvertPKIndex.class.getPackage().getImplementationVersion();\nif (!Objects.equals(v, plannerVersion)) throw new IllegalStateException(\"Mixed connector versions: \" + v + \" vs \" + plannerVersion);","typeGuard":"boolean isSupportedCommand(ReadCommand cmd) {\n  return cmd != null && (cmd.type() == CommandType.DATA || cmd.type() == CommandType.DELETE);\n}","tryCatchPattern":"// operator already routes to error stream; on consumer side\nDataStream<Throwable> errors = result.getErrorStream();\nerrors.process((ctx, t) -> {\n  if (msg.contains(\"Unexpected command type\")) alertMixedVersions();\n});","preventionTips":["Ship a single shaded Iceberg Flink connector version; check for duplicate classes on the classpath.","Add a job-startup version check comparing planner and operator artifacts.","Never inject ReadCommands into the keyed stream from custom code.","Test planner->PKIndex command flows in CI after any upgrade."],"tags":["flink","state-machine","internal-error"],"backgroundTag":"internal-invariant-violation","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"}