{"record":{"id":"96a843279c5d7872","repo":"apache/iceberg","slug":"invalid-operator-event-type-96a843","errorCode":null,"errorMessage":"Invalid operator event type: ","messagePattern":"Invalid operator event type: ","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManagerOperator.java#L201-L237","documentation":"TriggerManagerOperator.handleOperatorEvent only accepts LockReleaseEvent; any other OperatorEvent delivered to the maintenance task operator cannot be processed and is rejected with this IllegalArgumentException. It is the operator-side counterpart of the coordinator's event validation.","triggerScenarios":"A non-LockReleaseEvent (e.g. CheckpointTriggerEvent, TimerTriggerEvent, or a LockRegisterEvent) is sent to the TriggerManagerOperator instead of the coordinator, or version-skewed event classes arrive after a job restore.","commonSituations":"Custom code or tooling addressing events to the wrong operator; savepoint/restore across different Iceberg versions where the event class set changed; event routing bugs in custom maintenance topologies.","solutions":["Route trigger events and lock registrations to the TriggerManagerCoordinator, not the operator; only LockReleaseEvent belongs at the operator.","Use matching Iceberg Flink versions for savepoint production and restore.","Verify event wiring in any custom topology built with IcebergFlinkMaintenance builder APIs.","If running a mixed-version cluster, restart the whole job on a single version."],"exampleFix":"// before\ntaskOperator.sendOperatorEvent(new TimerTriggerEvent(cronSchedule));\n// after\ntaskCoordinator.sendOperatorEvent(new TimerTriggerEvent(cronSchedule));","handlingStrategy":"type-guard","validationCode":"if (!(event instanceof LockReleaseEvent)) {\n  throw new IllegalArgumentException(\"Event not valid for operator: \" + event.getClass());\n}","typeGuard":"boolean isOperatorEvent(OperatorEvent e) {\n  return e instanceof LockReleaseEvent;\n}","tryCatchPattern":"try { operator.sendOperatorEvent(event); } catch (IllegalArgumentException e) { LOG.warn(\"Wrong endpoint for event {}\", event.getClass(), e); }","preventionTips":["Only send LockReleaseEvent instances to TriggerManagerOperator.","Keep event-emitting code centralized in one helper that enforces the correct target.","Verify savepoint/restore use the same Iceberg Flink version that produced them."],"tags":["flink","operator-event","maintenance","type-mismatch"],"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"}