From 1398c4e65e4d699f09f1b5fc09ff000ac6e0558a Mon Sep 17 00:00:00 2001 From: rosemaryYuan <91046107+rosemarYuan@users.noreply.github.com> Date: Tue, 30 Jun 2026 18:04:24 +0800 Subject: [PATCH 1/4] [api][plan][runtime][python] Implement CEL-based @Action trigger condition filtering --- .../apache/flink/agents/api/EventType.java | 25 ++ .../flink/agents/api/annotation/Action.java | 30 +- .../api/configuration/AgentConfigOptions.java | 21 ++ docs/content/docs/operations/configuration.md | 1 + plan/pom.xml | 5 + .../apache/flink/agents/plan/AgentPlan.java | 61 +++- .../flink/agents/plan/actions/Action.java | 82 +++++- .../plan/condition/ConditionSyntaxPolicy.java | 206 +++++++++++++ .../plan/condition/ParsedCondition.java | 200 +++++++++++++ pom.xml | 1 + python/flink_agents/api/core_options.py | 17 ++ .../api/tests/test_core_options.py | 2 +- runtime/pom.xml | 13 + .../runtime/condition/ActionRouter.java | 129 +++++++++ .../runtime/condition/ConditionEvaluator.java | 273 ++++++++++++++++++ .../ConditionExpressionCompiler.java | 215 ++++++++++++++ .../operator/ActionExecutionOperator.java | 2 +- .../agents/runtime/operator/EventRouter.java | 15 +- .../runtime/operator/EventRouterTest.java | 2 +- 19 files changed, 1270 insertions(+), 30 deletions(-) create mode 100644 plan/src/main/java/org/apache/flink/agents/plan/condition/ConditionSyntaxPolicy.java create mode 100644 plan/src/main/java/org/apache/flink/agents/plan/condition/ParsedCondition.java create mode 100644 runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java create mode 100644 runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java create mode 100644 runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionExpressionCompiler.java diff --git a/api/src/main/java/org/apache/flink/agents/api/EventType.java b/api/src/main/java/org/apache/flink/agents/api/EventType.java index f486a6d0e..c12b2fbf0 100644 --- a/api/src/main/java/org/apache/flink/agents/api/EventType.java +++ b/api/src/main/java/org/apache/flink/agents/api/EventType.java @@ -18,6 +18,12 @@ package org.apache.flink.agents.api; +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + /** * Compile-time constants for built-in event types, sourced from each {@code XxxEvent.EVENT_TYPE}. * @@ -40,5 +46,24 @@ public final class EventType { public static final String ContextRetrievalResponseEvent = org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE; + /** + * Returns all built-in constants as an unmodifiable {@code name → event-type value} map. + * Enumerated reflectively from the {@code public static final String} fields of this class so + * newly added constants are picked up automatically. Iteration order is unspecified. + */ + public static Map allConstants() { + Map constants = new LinkedHashMap<>(); + for (Field field : EventType.class.getFields()) { + if (Modifier.isStatic(field.getModifiers()) && field.getType() == String.class) { + try { + constants.put(field.getName(), (String) field.get(null)); + } catch (IllegalAccessException e) { + throw new IllegalStateException("Cannot read EventType." + field.getName(), e); + } + } + } + return Collections.unmodifiableMap(constants); + } + private EventType() {} } diff --git a/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java b/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java index 49084255b..baad0d3c9 100644 --- a/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java +++ b/api/src/main/java/org/apache/flink/agents/api/annotation/Action.java @@ -26,9 +26,10 @@ /** * Marks a method as an agent action triggered by matching events. * - *

Each {@link #value()} entry is an event type name string. Use the {@code EVENT_TYPE} constants - * on built-in event classes, the {@link org.apache.flink.agents.api.EventType} constants, or plain - * strings for custom events. Multiple entries combine with OR. + *

Each {@link #value()} entry is an event type name string or a boolean condition expression. + * Use the {@code EVENT_TYPE} constants on built-in event classes, the {@link + * org.apache.flink.agents.api.EventType} constants, or plain strings for custom events. Multiple + * entries combine with OR. * * 

{@code
  * // Built-in event type via the EventType constant
@@ -40,8 +41,25 @@
  * // User-defined event type
  * @Action("MyCustomEvent")
  *
+ * // Event type containing '.' or '-': quote it
+ * @Action("'com.example.order-created'")
+ *
  * // Multiple types (OR semantics)
  * @Action({EventType.InputEvent, "MyCustomEvent"})
+ *
+ * // Boolean condition expressions, matched against the event's type and attributes:
+ *
+ * // Type match plus an attribute comparison
+ * @Action("type == EventType.InputEvent && score > 5")
+ *
+ * // Quoted type literal for a custom event containing '.' or '-'
+ * @Action("type == 'com.ios.orderEvent' && score > 10")
+ *
+ * // Attribute key containing '.' or '-': index access, then a field
+ * @Action("score_name['batch-3.12'].region > 10")
+ *
+ * // Test a special-character key with 'in' before indexing it
+ * @Action("'batch-3.12' in score_name && score_name['batch-3.12'].region > 10")
  * }
* *

For a cross-language action, set {@link #target()} to a {@link PythonFunction} with a @@ -61,9 +79,9 @@ @Retention(RetentionPolicy.RUNTIME) public @interface Action { /** - * Event type name strings; multiple entries have OR semantics. Named {@code value} (not {@code - * triggerConditions}) to enable the {@code @Action({...})} shorthand (JLS §9.7.3); corresponds - * to Python's {@code *trigger_conditions}. + * Event type name strings or boolean condition expressions; multiple entries have OR semantics. + * Named {@code value} (not {@code triggerConditions}) to enable the {@code @Action({...})} + * shorthand (JLS §9.7.3); corresponds to Python's {@code *trigger_conditions}. */ String[] value(); diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index c39997da1..c412951d6 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -25,6 +25,14 @@ /** The set of configuration options for agents parameters. */ public class AgentConfigOptions { + /** Behaviour when a trigger condition throws or returns a non-Boolean at evaluation time. */ + public enum ConditionEvaluationFailureStrategy { + /** Log WARN and treat the action as not matching. Streaming-safe default. */ + WARN_AND_SKIP, + /** Rethrow as {@code IllegalStateException}; triggers Flink task failover. */ + FAIL + } + /** * The config parameter specifies which event logger implementation to use. Defaults to {@link * LoggerType#SLF4J}, which surfaces events in Flink's Web UI; setting {@link LoggerType#FILE} @@ -33,6 +41,19 @@ public class AgentConfigOptions { public static final ConfigOption EVENT_LOGGER_TYPE = new ConfigOption<>("eventLoggerType", LoggerType.class, LoggerType.SLF4J); + /** + * Controls how trigger condition evaluation handles runtime exceptions and non-Boolean results. + * Defaults to {@link ConditionEvaluationFailureStrategy#WARN_AND_SKIP} for streaming safety; + * set to {@link ConditionEvaluationFailureStrategy#FAIL} on strict-semantics pipelines to + * trigger failover. + */ + public static final ConfigOption + CONDITION_EVALUATION_FAILURE_STRATEGY = + new ConfigOption<>( + "action.trigger-condition.evaluate-failure-strategy", + ConditionEvaluationFailureStrategy.class, + ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + /** The config parameter specifies the directory for the FileEvent file. */ public static final ConfigOption BASE_LOG_DIR = new ConfigOption<>("baseLogDir", String.class, null); diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index 44ebbc0fe..5744ad3bf 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -133,6 +133,7 @@ Here is the list of all built-in core configuration options. | `baseLogDir` | (none) | String | Base directory for file-based event logs. If not set, uses `java.io.tmpdir/flink-agents`. Setting this value also implicitly switches `eventLoggerType` to `file`. | | `prettyPrint` | false | boolean | Whether to enable pretty-printed JSON format for event logs. When set to `true`, each event is written as formatted multi-line JSON instead of JSONL (JSON Lines) format. {{< hint info >}}Note: enabling this option makes the log file no longer valid JSONL format. {{< /hint >}} | | `event-listeners` | none | `List` | The list of event listener class names. Each class must implement the EventListener interface and provide a public no-argument constructor. {{< hint warning >}} Note: Currently, custom event listeners are only supported in Java. {{< /hint >}} | +| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Strategy for handling a trigger-condition expression that throws or returns a non-boolean value at evaluation time.
The option value could be:

| | `error-handling-strategy` | ErrorHandlingStrategy.FAIL | ErrorHandlingStrategy | Strategy for handling errors during model requests, include timeout and unexpected output schema.
The option value could be: