{"record":{"id":"842620d891c3264c","repo":"apache/iceberg","slug":"invalid-operator-event-type-842620","errorCode":null,"errorMessage":"Invalid operator event type: ","messagePattern":"Invalid operator event type: ","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManagerOperator.java","lineNumber":219,"sourceCode":"      nextEvaluationTimeState.add(nextEvaluationTime);\n    }\n\n    accumulatedChangesState.update(accumulatedChanges);\n    lastTriggerTimesState.update(lastTriggerTimes);\n    LOG.info(\n        \"Storing state: nextEvaluationTime {}, accumulatedChanges {}, lastTriggerTimes {}\",\n        nextEvaluationTime,\n        accumulatedChanges,\n        lastTriggerTimes);\n  }\n\n  @Override\n  public void handleOperatorEvent(OperatorEvent event) {\n    if (event instanceof LockReleaseEvent) {\n      LOG.info(\"Received lock released event: {}\", event);\n      handleLockRelease((LockReleaseEvent) event);\n    } else {\n      throw new IllegalArgumentException(\n          \"Invalid operator event type: \" + event.getClass().getCanonicalName());\n    }\n  }\n\n  @Override\n  public void processElement(StreamRecord<TableChange> streamRecord) throws Exception {\n    TableChange change = streamRecord.getValue();\n    accumulatedChanges.forEach(tableChange -> tableChange.merge(change));\n    if (nextEvaluationTime == null) {\n      checkAndFire(getProcessingTimeService());\n    } else {\n      LOG.info(\n          \"Trigger manager rate limiter triggered current: {}, next: {}, accumulated changes: {},{}\",\n          getProcessingTimeService().getCurrentProcessingTime(),\n          nextEvaluationTime,\n          accumulatedChanges,\n          maintenanceTaskNames);\n      rateLimiterTriggeredCounter.inc();","sourceCodeStart":201,"sourceCodeEnd":237,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManagerOperator.java#L201-L237","documentation":"TriggerManagerOperator.handleOperatorEvent() only processes LockReleaseEvent sent from its coordinator. Any other OperatorEvent type indicates a protocol mismatch between coordinator and operator, so it throws IllegalArgumentException with the event's class name.","triggerScenarios":"The TriggerManagerCoordinator sends an event type other than LockReleaseEvent (e.g. LockRegisterEvent) to this operator; custom code invokes handleOperatorEvent with a foreign OperatorEvent.","commonSituations":"Mismatched coordinator/operator versions after a savepoint restore or partial upgrade; custom operators reusing this operator's event gateway; tests invoking handleOperatorEvent with wrong event instances.","solutions":["Verify the coordinator paired with this operator only sends LockReleaseEvent","Align Iceberg versions across all job vertices (avoid partial upgrades / mixed classpath)","Check that the job was not restored from a savepoint produced by an incompatible version","If new event types are needed, add instanceof branches to handleOperatorEvent"],"exampleFix":"// before\noperator.handleOperatorEvent(new LockRegisterEvent(factory, id)); // IllegalArgumentException\n// after\ncoordinatorContext.sendOperatorEvent(new LockReleaseEvent(id));","handlingStrategy":"type-guard","validationCode":"// before invoking the operator's event handler\nif (!(event instanceof LockReleaseEvent)) {\n  throw new IllegalArgumentException(\"TriggerManagerOperator only accepts LockReleaseEvent, got \" + event.getClass());\n}","typeGuard":"if (event instanceof LockReleaseEvent release) {\n  operator.handleOperatorEvent(release);\n} else { /* route to the right operator */ }","tryCatchPattern":"try {\n  operator.handleOperatorEvent(event);\n} catch (IllegalArgumentException e) {\n  LOG.error(\"Coordinator sent unsupported OperatorEvent to TriggerManagerOperator\", e);\n  throw e;\n}","preventionTips":["Keep coordinator and operator from the same Iceberg version (no mixed classpath)","Register the coordinator only for the operator it was built for","Review custom event-sending code before wiring it into the maintenance job"],"tags":["flink","operator-event","protocol-mismatch","illegal-argument"],"backgroundTag":"invalid-argument-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"}