{"record":{"id":"0319f17b761f562d","repo":"apache/iceberg","slug":"received-unknown-event-from-subtask-d-s-0319f1","errorCode":null,"errorMessage":"Received unknown event from subtask %d: %s","messagePattern":"Received unknown event from subtask (.+?): (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/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/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/AbstractIcebergEnumerator.java#L77-L113","documentation":"AbstractIcebergEnumerator.handleSourceEvent expects either a SplitRequestEvent (finished split ids + hostname) or an AddSplitBackEvent (waitLikeContinuous). Any other SourceEvent arriving from a subtask is unknown to the Iceberg protocol, and the enumerator throws IllegalArgumentException with the event's canonical class name and the subtask id. This guards against reader/enumerator protocol mismatches.","triggerScenarios":"A reader sends a custom SourceEvent class (or a version-mismatched event type) to an Iceberg enumerator — e.g., a custom implementation of IcebergSourceReader emitting its own event, or mixed Iceberg versions across the cluster where event classes differ.","commonSituations":"Deploying mismatched Iceberg-flink reader/enumerator jars in the same job after an upgrade; custom extensions of the source that add new event types without extending the enumerator; fat-jar shading producing duplicate event classes.","solutions":["Ensure all TaskManagers run the same iceberg-flink runtime version — re-build/redeploy a consistent job jar.","Check for custom SourceEvent subclasses in your code; route them through the enumerator or convert them to SplitRequestEvent.","Inspect the class name in the message to identify which component sent the unknown event and fix that sender.","If you extended the protocol, override handleSourceEvent to accept your event before falling back to super."],"exampleFix":"// before\n@Override\npublic void handleSourceEvent(int subtaskId, SourceEvent event) { super.handleSourceEvent(subtaskId, event); }\n// after\n@Override\npublic void handleSourceEvent(int subtaskId, SourceEvent event) {\n  if (event instanceof MyCustomEvent) { handleCustom(subtaskId, (MyCustomEvent) event); return; }\n  super.handleSourceEvent(subtaskId, event);\n}","handlingStrategy":"validation","validationCode":"if (!(event instanceof SplitRequestEvent) && !(event instanceof AddSplitBackEvent)) {\n  throw new IllegalArgumentException(\"Unsupported SourceEvent for Iceberg enumerator: \" + event.getClass());\n}","typeGuard":"boolean isIcebergSourceEvent(SourceEvent e) { return e instanceof SplitRequestEvent || e instanceof AddSplitBackEvent; }","tryCatchPattern":null,"preventionTips":["Deploy identical iceberg-flink jars on all nodes to avoid event class mismatch.","If adding custom SourceEvents, extend the enumerator's handleSourceEvent to accept them.","Avoid shading that duplicates SourceEvent classes under different classloaders."],"tags":["flink","source-enumerator","protocol-mismatch","version-skew"],"backgroundTag":"unexpected-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"}