{"record":{"id":"63c1f62d81fc5888","repo":"apache/iceberg","slug":"received-unknown-event-from-subtask-d-s","errorCode":null,"errorMessage":"Received unknown event from subtask %d: %s","messagePattern":"Received unknown event from subtask (.+?): (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/AbstractIcebergEnumerator.java","lineNumber":95,"sourceCode":"    // Iceberg source uses custom split request event to piggyback finished split ids.\n    throw new UnsupportedOperationException(\n        String.format(\n            Locale.ROOT,\n            \"Received invalid default split request event \"\n                + \"from subtask %d as Iceberg source uses custom split request event\",\n            subtaskId));\n  }\n\n  @Override\n  public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {\n    if (sourceEvent instanceof SplitRequestEvent) {\n      SplitRequestEvent splitRequestEvent = (SplitRequestEvent) sourceEvent;\n      LOG.info(\"Received request split event from subtask {}\", subtaskId);\n      assigner.onCompletedSplits(splitRequestEvent.finishedSplitIds());\n      readersAwaitingSplit.put(subtaskId, splitRequestEvent.requesterHostname());\n      assignSplits();\n    } else {\n      throw new IllegalArgumentException(\n          String.format(\n              Locale.ROOT,\n              \"Received unknown event from subtask %d: %s\",\n              subtaskId,\n              sourceEvent.getClass().getCanonicalName()));\n    }\n  }\n\n  // Flink's SourceCoordinator already keeps track of subTask to splits mapping.\n  // It already takes care of re-assigning splits to speculated attempts as well.\n  @Override\n  public void handleSourceEvent(int subTaskId, int attemptNumber, SourceEvent sourceEvent) {\n    handleSourceEvent(subTaskId, sourceEvent);\n  }\n\n  @Override\n  public void addSplitsBack(List<IcebergSourceSplit> splits, int subtaskId) {\n    LOG.info(\"Add {} splits back to the pool for failed subtask {}\", splits.size(), subtaskId);","sourceCodeStart":77,"sourceCodeEnd":113,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/AbstractIcebergEnumerator.java#L77-L113","documentation":"The IcebergSource enumerator's handleSourceEvent received a SourceEvent from a reader subtask that it does not recognize. It only understands SplitRequestEvent; anything else is rejected. This typically indicates an internal protocol mismatch between reader and enumerator (e.g., mixed Iceberg connector versions on the same job).","triggerScenarios":"A SourceReader sends a custom/unknown SourceEvent to the enumerator via the split enumerator context, and handleSourceEvent is called with an event that is not a SplitRequestEvent.","commonSituations":"Running mismatched Iceberg Flink connector jar versions between JobManager and TaskManagers; custom reader subclasses emitting their own events; classpath contamination with two connector versions.","solutions":["Ensure all TaskManagers and the JobManager use the exact same Iceberg Flink connector version.","Check for multiple iceberg-flink jars on the classpath and remove duplicates from the user jar (shade/relocate).","If a custom reader emits extra events, override handleSourceEvent to handle them before falling through to the default branch."],"exampleFix":"// before\nsourceReaderContext.sendSourceEvent(myCustomEvent);\n// after\n// only send SplitRequestEvent, or handle custom events in an enumerator subclass:\n@Override\npublic void handleSourceEvent(int subtaskId, SourceEvent event) {\n  if (event instanceof MyCustomEvent) { handleCustom((MyCustomEvent) event); return; }\n  super.handleSourceEvent(subtaskId, event);\n}","handlingStrategy":"validation","validationCode":"// Ensure consistent connector version at deployment time\nString expected = \"org.apache.iceberg:iceberg-flink-runtime-1.20:<version>\";\nString found = org.apache.iceberg.flink.IcebergFlinkVersion.class.getPackage().getImplementationVersion();\nif (!expected.endsWith(found)) throw new IllegalStateException(\"Connector version mismatch: \" + found);","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Deploy one iceberg-flink jar version across the whole cluster","Shade/relocate the connector inside user jars consistently","Only send SplitRequestEvent from readers unless extending the enumerator"],"tags":["flink","source-enumerator","protocol-mismatch","illegal-argument"],"backgroundTag":"unexpected-api-response-shape","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"}