[plan][runtime][java] Add CEL Action condition filtering

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,25 @@ 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<String, String> allConstants() {
+        Map<String, String> 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) {
+                    // Unreachable: getFields() only returns public fields.
+                    throw new IllegalStateException("Cannot read EventType." + field.getName(), e);
+                }
+            }
+        }
+        return Collections.unmodifiableMap(constants);
+    }
+
     private EventType() {}
 }

api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java

@@ -33,6 +33,17 @@ public class AgentConfigOptions {
     public static final ConfigOption<LoggerType> EVENT_LOGGER_TYPE =
             new ConfigOption<>("eventLoggerType", LoggerType.class, LoggerType.SLF4J);
 
+    /**
+     * Controls how the CEL condition evaluator handles runtime exceptions and non-Boolean results.
+     * Defaults to {@link CelEvaluationFailurePolicy#WARN_AND_SKIP} for streaming safety; set to
+     * {@link CelEvaluationFailurePolicy#FAIL} on strict-semantics pipelines to trigger failover.
+     */
+    public static final ConfigOption<CelEvaluationFailurePolicy> CEL_EVALUATION_FAILURE_POLICY =
+            new ConfigOption<>(
+                    "celEvaluationFailurePolicy",
+                    CelEvaluationFailurePolicy.class,
+                    CelEvaluationFailurePolicy.WARN_AND_SKIP);
+
     /** The config parameter specifies the directory for the FileEvent file. */
     public static final ConfigOption<String> BASE_LOG_DIR =
             new ConfigOption<>("baseLogDir", String.class, null);

api/src/main/java/org/apache/flink/agents/api/configuration/CelEvaluationFailurePolicy.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.api.configuration;
+
+/** Behaviour when a CEL trigger condition throws or returns a non-Boolean at evaluation time. */
+public enum CelEvaluationFailurePolicy {
+    /** Log WARN and treat the action as not matching. Streaming-safe default. */
+    WARN_AND_SKIP,
+    /** Rethrow as {@code IllegalStateException}; triggers Flink task failover. */
+    FAIL
+}

dist/src/main/resources/META-INF/NOTICE

@@ -14,6 +14,8 @@ This project bundles the following dependencies under the Apache Software Licens
 - com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.2
 - com.fasterxml.jackson.module:jackson-module-kotlin:2.18.2
 - com.fasterxml:classmate:1.7.0
+- dev.cel:cel:0.12.0
+- dev.cel:protobuf:0.12.0
 - org.apache.logging.log4j:log4j-api:2.23.1
 - org.apache.logging.log4j:log4j-core:2.23.1
 - org.apache.logging.log4j:log4j-slf4j-impl:2.23.1
@@ -182,6 +184,8 @@ This project bundles the following dependencies under the BSD 3-Clause license.
 See bundled license files for details.
 
 - com.google.protobuf:protobuf-java:3.25.5
+- com.google.re2j:re2j:1.8
+- org.antlr:antlr4-runtime:4.13.2
 - org.ow2.asm:asm:9.3
 
 This project bundles the following dependencies under the EPL2 license.

dist/src/main/resources/META-INF/licenses/LICENSE.antlr4-runtime

@@ -0,0 +1,28 @@
+Copyright (c) 2012-2022 The ANTLR Project. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+1. Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright
+notice, this list of conditions and the following disclaimer in the
+documentation and/or other materials provided with the distribution.
+
+3. Neither name of copyright holders nor the names of its contributors
+may be used to endorse or promote products derived from this software
+without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR
+CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

dist/src/main/resources/META-INF/licenses/LICENSE.re2j

@@ -0,0 +1,32 @@
+This is a work derived from Russ Cox's RE2 in Go, whose license
+http://golang.org/LICENSE is as follows:
+
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+     notice, this list of conditions and the following disclaimer.
+
+   * Redistributions in binary form must reproduce the above copyright
+     notice, this list of conditions and the following disclaimer in
+     the documentation and/or other materials provided with the
+     distribution.
+
+   * Neither the name of Google Inc. nor the names of its contributors
+     may be used to endorse or promote products derived from this
+     software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.