wenjin272 commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3552878422
########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "Trigger condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the variable map used to evaluate a trigger condition for one event. It always + * contains: + * + * <ul> + * <li>{@code type}: the event routing type, from {@link Event#getType()}. + * <li>{@code id}: the user-provided {@code id} attribute when present; otherwise the event + * UUID. + * <li>{@code attributes}: the normalized event attributes. + * <li>{@code EventType}: built-in event type constants, e.g. {@code EventType.InputEvent}. + * </ul> + * + * <p>Event attributes are also exposed as top-level variables: an attribute {@code score} can + * be referenced as either {@code score} or {@code attributes.score}. + * + * <p>Immediate keys of nested {@code input}/{@code output} maps are exposed as top-level + * attributes too; on collision, {@code output} keys win over root attributes, which win over + * {@code input} keys. + * + * <p>String values that look like JSON objects or arrays are parsed before evaluation; narrow + * numeric types widen to {@code long} or {@code double}. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> buildTriggerVariables(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); + Map<String, Object> merged = new HashMap<>(); + if (normalizedAttrs instanceof Map) { + Map<String, Object> attrs = (Map<String, Object>) normalizedAttrs; + + // Precedence: output subkeys > root attributes > input subkeys (putIfAbsent keeps the + // earliest insertion). Root iteration includes the "input"/"output" maps themselves, + // so nested paths like input.region.width keep working. + Object outputObj = attrs.get("output"); + if (outputObj instanceof Map) { + ((Map<String, Object>) outputObj).forEach(merged::putIfAbsent); + } + attrs.forEach(merged::putIfAbsent); + Object inputObj = attrs.get("input"); + if (inputObj instanceof Map) { + ((Map<String, Object>) inputObj).forEach(merged::putIfAbsent); + } + } + + activation.put("attributes", merged); + // Promote to top level for bare-identifier access; framework keys win on collision. + merged.forEach(activation::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + activation.putIfAbsent("id", event.getId().toString()); + + return activation; + } + + /** + * Builds a trimmed activation covering the union of {@code conditions}' referenced merged keys. + * Returns {@code null} under {@link ConditionEvaluationFailureStrategy#WARN_AND_SKIP} when the + * build fails, so the event's trigger conditions are skipped; rethrows as {@link + * IllegalStateException} under {@link ConditionEvaluationFailureStrategy#FAIL}. Any condition + * whose key set was not pre-compiled forces the full (untrimmed) build for safety. + */ + @Nullable + public Map<String, Object> buildTriggerVariablesForConditions( Review Comment: Comment 1: simplify the variable-building paths I think the full-build / partial-build fallback structure is adding more complexity than the trigger condition semantics require. If we want to keep the referenced-key optimization, the runtime could have a single main path: 1. collect the referenced attribute keys from the precompiled condition metadata, 2. build the CEL activation only for those keys, 3. add the framework-owned variables such as `type`, `EventType`, and the event-id fallback. The current fallback paths mostly exist because the full build supports extra implicit behavior (`attributes` as a full namespace, JSON-looking strings being parsed, and `input`/ `output` subkeys being flattened). If those semantics are not part of the intended public condition DSL, we should not preserve them through fallback logic. Concretely, `buildTriggerVariablesForConditions` can stay as the main entry point, with maybe one helper for collecting referenced keys. `buildTriggerVariables` / `buildPartialTriggerVariables` do not need to be separate public-style paths. Comment 2: avoid exposing attributes as a special framework variable I find the special `attributes` variable confusing here. `Event.attributes` is already the user payload. For action conditions, the natural user-facing model should be that keys inside `event.getAttributes()` are exposed as condition variables, e.g. `score > 80`, not that users need to know about the Java `Event` object shape and write `attributes.score`. Supporting both `score` and `attributes.score` creates a second access model and forces special handling in `extractMergedKeys()` (`attributes` as a whole-map sentinel). It also makes `attributes` a reserved name, which can conflict with a real user attribute named `attributes`. I suggest not injecting a framework-owned full `attributes` map into the CEL activation. Treat `attributes` as just another user attribute key if the user provided one, and only reserve explicit framework variables such as `type` and `EventType`. Comment 3: remove JSON-string parsing from normalization I do not think `normalizeValue` should parse JSON-looking strings. The numeric normalization makes sense because CEL expects stable numeric types, but parsing arbitrary strings that look like JSON changes the user payload type implicitly. For example, a string value like `"{"region":15}"` stops being a string and becomes a map only because of its shape. That makes condition behavior harder to reason about and creates special cases in the referenced-key optimization. If `input` or `output` should be structured, they should be provided as `Map` values. If they are strings, they should remain strings. With that rule, the JSON-string fallback in `buildPartialTriggerVariables` is no longer needed. Comment 4: fail fast on missing referenced-key metadata The `keySetCache == null` / cache-miss fallback to full variable building seems too permissive. In the normal `ActionRouter` lifecycle, expression conditions should be initialized through `initPrograms()`, and the program cache and referenced-key cache should be built together. If referenced-key metadata is missing for a condition, that looks like a lifecycle/invariant violation rather than a case where we should silently switch to full activation building. Since this PR is adding referenced-key optimization, I think missing key metadata should fail fast with a clear error. That keeps the optimized path easier to reason about and avoids keeping a second full-build behavior only as a defensive fallback. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "Trigger condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the variable map used to evaluate a trigger condition for one event. It always + * contains: + * + * <ul> + * <li>{@code type}: the event routing type, from {@link Event#getType()}. + * <li>{@code id}: the user-provided {@code id} attribute when present; otherwise the event + * UUID. + * <li>{@code attributes}: the normalized event attributes. + * <li>{@code EventType}: built-in event type constants, e.g. {@code EventType.InputEvent}. + * </ul> + * + * <p>Event attributes are also exposed as top-level variables: an attribute {@code score} can + * be referenced as either {@code score} or {@code attributes.score}. + * + * <p>Immediate keys of nested {@code input}/{@code output} maps are exposed as top-level + * attributes too; on collision, {@code output} keys win over root attributes, which win over + * {@code input} keys. + * + * <p>String values that look like JSON objects or arrays are parsed before evaluation; narrow + * numeric types widen to {@code long} or {@code double}. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> buildTriggerVariables(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); + Map<String, Object> merged = new HashMap<>(); + if (normalizedAttrs instanceof Map) { + Map<String, Object> attrs = (Map<String, Object>) normalizedAttrs; + + // Precedence: output subkeys > root attributes > input subkeys (putIfAbsent keeps the + // earliest insertion). Root iteration includes the "input"/"output" maps themselves, + // so nested paths like input.region.width keep working. + Object outputObj = attrs.get("output"); + if (outputObj instanceof Map) { Review Comment: The `input` / `output` subkey promotion seems too broad here. This code promotes subkeys whenever the event attributes contain `input` or `output` maps, regardless of the actual event type. If this is intended as a convenience for framework `InputEvent` / `OutputEvent`, it should probably be scoped to those event types only. For ordinary user events, `input` and`output` should remain normal attribute keys, and users can access nested values as `input.foo` / `output.foo`. Please also update `resolveMergedKey()` consistently. It currently resolves every referenced key by checking `output.<key>` first, then the root attribute, then `input.<key>`, for all event types. If subkey promotion is limited to `InputEvent` / `OutputEvent`, this lookup order should only apply to the matching framework event type; otherwise a user event with an `input` or `output` map can still have its nested fields promoted in the optimized path. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { Review Comment: Do we need `initPrograms()` as a separate lifecycle method? In the production path it is called immediately after constructing `ConditionEvaluator`, so these two steps seem to define one complete evaluator instance. Keeping them separate leaves the object in a partially initialized state, forces `programCache` / `keySetCache` to be nullable, and requires extra defensive handling for missing cache state. It may be simpler to pass the expression conditions into the constructor or a factory method, build both caches there, and make them final. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. Review Comment: Can we avoid mentioning CEL in this class-level Javadoc and related comments/variable names? `ActionRouter` is describing runtime trigger matching, while CEL is only the current implementation used for expression conditions. For example, this could be phrased in domain terms such as “exact event-type conditions” and “expression conditions” instead of “type-index fast path” and “CEL slow path”. That would make the API easier to read and avoid using “CEL” inconsistently to mean the language, SDK, compiled program cache, or evaluation path. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; + } + conditionEvaluator = + new ConditionEvaluator( + agentPlan + .getConfig() + .get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY)); + conditionEvaluator.initPrograms(expressionConditions); + } + + /** Returns actions to fire for {@code event}: typed hits first, then CEL hits. */ Review Comment: ``` /** * Returns the actions whose trigger conditions match the event. * * <p>Actions matched by exact event type are returned before actions matched by expression * conditions. Each action appears at most once. */ ``` ########## plan/src/main/java/org/apache/flink/agents/plan/condition/ParsedCondition.java: ########## @@ -0,0 +1,200 @@ +/* + * 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.plan.condition; + +import dev.cel.common.CelAbstractSyntaxTree; +import dev.cel.common.CelValidationException; +import dev.cel.common.ast.CelConstant; +import dev.cel.common.ast.CelExpr; + +import java.util.Objects; + +/** + * A parsed {@code Action.triggerConditions} entry — either {@link TypeMatch} or {@link + * ExpressionCondition}. {@link #classify} turns a raw entry string into one of the two. + */ +public interface ParsedCondition { + + /** + * Matching key: the resolved event-type name for {@link TypeMatch}, or the expression source + * for {@link ExpressionCondition}. + */ + String source(); + + /** + * Classifies a {@code triggerConditions} entry. A bare identifier or a quoted string is an + * event-type name ({@link TypeMatch}); anything else must be a boolean condition expression + * ({@link ExpressionCondition}). Name-shaped entries that are neither — e.g. {@code a.b.event} + * or {@code order-created} written without quotes — are rejected with the quoted spelling + * suggested. + */ + static ParsedCondition classify(String source) { Review Comment: I wonder if we should split exact event-type triggers and expression triggers in the API instead of overloading one `value` / `triggerConditions` field for both. Today the same string list can contain either event type names or boolean expressions, which forces `ParsedCondition.classify()` to guess the intent, introduces special rules for name-shaped entries and quoted event types, and makes the plan module depend on CEL parsing just to classify trigger entries. A cleaner API could be something like `value` for exact event types and `conditions` for expression triggers. Then event types never need to be parsed as expressions, complex event type names do not need nested quoting like `"'com.example.order-created'"`, and CEL-specific parsing can stay in the runtime expression compiler. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionExpressionCompiler.java: ########## @@ -0,0 +1,270 @@ +/* + * 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.runtime.condition; + +import dev.cel.common.CelAbstractSyntaxTree; +import dev.cel.common.CelOptions; +import dev.cel.common.CelValidationException; +import dev.cel.common.ast.CelConstant; +import dev.cel.common.ast.CelExpr; +import dev.cel.common.navigation.CelNavigableAst; +import dev.cel.common.navigation.CelNavigableExpr; +import dev.cel.common.types.CelType; +import dev.cel.common.types.MapType; +import dev.cel.common.types.SimpleType; +import dev.cel.compiler.CelCompiler; +import dev.cel.compiler.CelCompilerBuilder; +import dev.cel.compiler.CelCompilerFactory; +import dev.cel.parser.CelParser; +import dev.cel.parser.CelParserFactory; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import dev.cel.runtime.CelRuntimeFactory; +import org.apache.flink.agents.api.EventType; +import org.apache.flink.agents.plan.condition.ConditionSyntaxPolicy; +import org.apache.flink.annotation.VisibleForTesting; + +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** Compiles trigger condition expressions into cached executable programs. */ +public final class ConditionExpressionCompiler { + + /** Immutable {@code constant name → value} map bound to the CEL {@code EventType} variable. */ + static final Map<String, Object> EVENT_TYPE_CONSTANTS = Map.copyOf(EventType.allConstants()); + + /** These caps bound expression size and depth. */ + private static final CelOptions CEL_OPTIONS = + CelOptions.current() + .maxExpressionCodePointSize(8_192) + .maxParseRecursionDepth(32) + .comprehensionMaxIterations(1_000) + .build(); + + /** + * Only the custom has() macro is enabled; all others are rejected by {@link + * ConditionSyntaxPolicy}. + */ + private static final CelParser CEL_PARSER = + CelParserFactory.standardCelParserBuilder() + .setOptions(CEL_OPTIONS) + .addMacros(ConditionSyntaxPolicy.HAS) + .build(); + + /** Vars always declared at type-check. */ + private static final Map<String, CelType> BASE_VARS = + Map.of( + "type", + SimpleType.STRING, + "id", + SimpleType.DYN, + "EventType", + MapType.create(SimpleType.STRING, SimpleType.STRING), + "attributes", + MapType.create(SimpleType.STRING, SimpleType.DYN)); + + private static final CelRuntime CEL_RUNTIME = + CelRuntimeFactory.standardCelRuntimeBuilder().setOptions(CEL_OPTIONS).build(); + + /** Process-wide bounded LRU cache of compiled CEL programs, keyed by source string. */ + static final int PROGRAM_CACHE_MAX_SIZE = 1024; + + private static final Map<String, CelRuntime.Program> PROGRAM_CACHE = Review Comment: Do we need the process-wide `PROGRAM_CACHE` in `ConditionExpressionCompiler` `ConditionEvaluator` already builds a per-evaluator `programCache` during initialization, so the route hot path does not recompile expressions. The global cache only adds reuse across different evaluators/plans, but it also introduces mutable global state, LRU behavior, and test-only hooks (`clearProgramCacheForTests`, `programCacheSizeForTests`). Unless we have a clear cross-plan reuse requirement or performance evidence, it may be simpler to keep only the evaluator-level compiled program map and let `toProgram()` compile directly. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( Review Comment: Is the `evaluate()` / `evaluateProgram()` split needed? `evaluateProgram()` has only one caller and mainly contains the second half of the same evaluation flow. This makes the failure handling split across two methods: `evaluate()` handles `CelEvaluationException`, while `evaluateProgram()` handles non-boolean results. I think this would be easier to read if the program lookup, evaluation, result type check, and failure-policy handling stayed in `evaluate()`. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { Review Comment: Do we need a separate `open()` lifecycle method here? In the production path, `ActionRouter.open()` is called immediately after constructing `ActionRouter`, so these two steps effectively create one usable router instance. Keeping them separate leaves the router in a partially initialized state, where `conditionEvaluator` may be null depending on whether `open()` was called and whether expression conditions exist. It may be simpler to initialize the expression-condition evaluator in the constructor (or a factory), and avoid the extra mutable lifecycle state. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; Review Comment: Can we avoid the early `return` here and structure this as `if (!expressionConditions.isEmpty()) { ... }` instead? `open()` is an initialization method, so returning early leaves the initialized state implicit. Using a positive condition around the optional evaluator setup would make it clearer that the router is still opened, but there are simply no expression conditions to precompile. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { Review Comment: Suggest `ActionMatcher`, because it is actually the `Event` that gets routed. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "Trigger condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the variable map used to evaluate a trigger condition for one event. It always + * contains: + * + * <ul> + * <li>{@code type}: the event routing type, from {@link Event#getType()}. + * <li>{@code id}: the user-provided {@code id} attribute when present; otherwise the event + * UUID. + * <li>{@code attributes}: the normalized event attributes. + * <li>{@code EventType}: built-in event type constants, e.g. {@code EventType.InputEvent}. + * </ul> + * + * <p>Event attributes are also exposed as top-level variables: an attribute {@code score} can + * be referenced as either {@code score} or {@code attributes.score}. + * + * <p>Immediate keys of nested {@code input}/{@code output} maps are exposed as top-level + * attributes too; on collision, {@code output} keys win over root attributes, which win over + * {@code input} keys. + * + * <p>String values that look like JSON objects or arrays are parsed before evaluation; narrow + * numeric types widen to {@code long} or {@code double}. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> buildTriggerVariables(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); + Map<String, Object> merged = new HashMap<>(); + if (normalizedAttrs instanceof Map) { + Map<String, Object> attrs = (Map<String, Object>) normalizedAttrs; + + // Precedence: output subkeys > root attributes > input subkeys (putIfAbsent keeps the + // earliest insertion). Root iteration includes the "input"/"output" maps themselves, + // so nested paths like input.region.width keep working. + Object outputObj = attrs.get("output"); + if (outputObj instanceof Map) { + ((Map<String, Object>) outputObj).forEach(merged::putIfAbsent); + } + attrs.forEach(merged::putIfAbsent); + Object inputObj = attrs.get("input"); + if (inputObj instanceof Map) { + ((Map<String, Object>) inputObj).forEach(merged::putIfAbsent); + } + } + + activation.put("attributes", merged); + // Promote to top level for bare-identifier access; framework keys win on collision. + merged.forEach(activation::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + activation.putIfAbsent("id", event.getId().toString()); + + return activation; + } + + /** + * Builds a trimmed activation covering the union of {@code conditions}' referenced merged keys. + * Returns {@code null} under {@link ConditionEvaluationFailureStrategy#WARN_AND_SKIP} when the + * build fails, so the event's trigger conditions are skipped; rethrows as {@link + * IllegalStateException} under {@link ConditionEvaluationFailureStrategy#FAIL}. Any condition + * whose key set was not pre-compiled forces the full (untrimmed) build for safety. + */ + @Nullable + public Map<String, Object> buildTriggerVariablesForConditions( + Event event, Collection<ExpressionCondition> conditions) { + try { + Set<String> unionKeys = new HashSet<>(); + for (ExpressionCondition c : conditions) { + if (c == null) { + continue; + } + Set<String> keys = keySetCache == null ? null : keySetCache.get(c.source()); + if (keys == null) { + return buildTriggerVariables(event); + } + unionKeys.addAll(keys); + } + return buildPartialTriggerVariables(event, unionKeys); + } catch (RuntimeException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Building trigger condition variables failed for event " + event.getId(), + e); + } + LOG.warn( + "Building trigger condition variables failed for event {}, skipping conditions", + event.getId(), + e); + return null; + } + } + + /** + * Like {@link #buildTriggerVariables} but only materializes the merged keys in {@code + * unionKeys}, reproducing the same output>root>input precedence and normalization depth. Falls + * back to the full build when trimming is unsafe: the {@code "attributes"} sentinel is present, + * or a flattened {@code output}/{@code input} container is stored as a JSON string (which the + * full build would parse and flatten, hiding subkeys from the per-key probe). + */ + @SuppressWarnings("unchecked") + private Map<String, Object> buildPartialTriggerVariables(Event event, Set<String> unionKeys) { + if (unionKeys.contains("attributes")) { + return buildTriggerVariables(event); + } + Object rawAttrsObj = event.getAttributes(); + if (!(rawAttrsObj instanceof Map)) { + return buildTriggerVariables(event); + } + Map<String, Object> rawAttrs = (Map<String, Object>) rawAttrsObj; Review Comment: Why is it necessary to check and forcibly cast the return type of `event.getAttributes()` here? Isn't the return value of `event.getAttributes()` already a definite `Map<String, Object>`? ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; + } + conditionEvaluator = + new ConditionEvaluator( + agentPlan + .getConfig() + .get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY)); + conditionEvaluator.initPrograms(expressionConditions); + } + + /** Returns actions to fire for {@code event}: typed hits first, then CEL hits. */ + public List<Action> route(Event event) { + List<Action> typedHits = + agentPlan + .getActionsByEvent() + .getOrDefault(event.getType(), Collections.emptyList()); + + // CEL candidates = actions with at least one CEL entry, excluding those already + // matched by typed routing for this event type. This avoids double-firing. + List<Action> celCandidates; + List<Action> withCel = agentPlan.getActionsWithExpressions(); + if (withCel.isEmpty()) { + celCandidates = Collections.emptyList(); + } else { + celCandidates = new ArrayList<>(); + for (Action a : withCel) { + if (!typedHits.contains(a)) { + celCandidates.add(a); + } + } + } + + if (celCandidates.isEmpty()) { + return typedHits; + } + + // Preserves typed-first ordering and deduplicates. + LinkedHashSet<Action> matched = new LinkedHashSet<>(typedHits); + + Map<String, Object> activation = + conditionEvaluator.buildTriggerVariablesForConditions( + event, expressionConditionsOf(celCandidates)); + if (activation == null) { + return new ArrayList<>(matched); + } + for (Action a : celCandidates) { + // Within-action OR: first matching CEL expression admits the action. + for (ParsedCondition pc : a.getParsedConditions()) { + if (!(pc instanceof ExpressionCondition)) { + continue; + } + if (conditionEvaluator.evaluate((ExpressionCondition) pc, activation)) { + matched.add(a); + break; + } + } + } + return new ArrayList<>(matched); + } + + private static List<ExpressionCondition> expressionConditionsOf(Collection<Action> actions) { + List<ExpressionCondition> result = new ArrayList<>(); + for (Action action : actions) { + for (ParsedCondition pc : action.getParsedConditions()) { + if (pc instanceof ExpressionCondition) { + result.add((ExpressionCondition) pc); + } + } + } + return result; + } + + /** Idempotent. */ + public void close() { Review Comment: Is `ActionRouter.close()` needed? It only closes and clears the in-memory `ConditionEvaluator`, and `ConditionEvaluator.close()` itself does not release any external resource. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "Trigger condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the variable map used to evaluate a trigger condition for one event. It always + * contains: + * + * <ul> + * <li>{@code type}: the event routing type, from {@link Event#getType()}. + * <li>{@code id}: the user-provided {@code id} attribute when present; otherwise the event + * UUID. + * <li>{@code attributes}: the normalized event attributes. + * <li>{@code EventType}: built-in event type constants, e.g. {@code EventType.InputEvent}. + * </ul> + * + * <p>Event attributes are also exposed as top-level variables: an attribute {@code score} can + * be referenced as either {@code score} or {@code attributes.score}. + * + * <p>Immediate keys of nested {@code input}/{@code output} maps are exposed as top-level + * attributes too; on collision, {@code output} keys win over root attributes, which win over + * {@code input} keys. + * + * <p>String values that look like JSON objects or arrays are parsed before evaluation; narrow + * numeric types widen to {@code long} or {@code double}. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> buildTriggerVariables(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); + Map<String, Object> merged = new HashMap<>(); + if (normalizedAttrs instanceof Map) { + Map<String, Object> attrs = (Map<String, Object>) normalizedAttrs; + + // Precedence: output subkeys > root attributes > input subkeys (putIfAbsent keeps the + // earliest insertion). Root iteration includes the "input"/"output" maps themselves, + // so nested paths like input.region.width keep working. + Object outputObj = attrs.get("output"); + if (outputObj instanceof Map) { + ((Map<String, Object>) outputObj).forEach(merged::putIfAbsent); + } + attrs.forEach(merged::putIfAbsent); + Object inputObj = attrs.get("input"); + if (inputObj instanceof Map) { + ((Map<String, Object>) inputObj).forEach(merged::putIfAbsent); + } + } + + activation.put("attributes", merged); + // Promote to top level for bare-identifier access; framework keys win on collision. + merged.forEach(activation::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + activation.putIfAbsent("id", event.getId().toString()); + + return activation; + } + + /** + * Builds a trimmed activation covering the union of {@code conditions}' referenced merged keys. + * Returns {@code null} under {@link ConditionEvaluationFailureStrategy#WARN_AND_SKIP} when the + * build fails, so the event's trigger conditions are skipped; rethrows as {@link + * IllegalStateException} under {@link ConditionEvaluationFailureStrategy#FAIL}. Any condition + * whose key set was not pre-compiled forces the full (untrimmed) build for safety. + */ + @Nullable + public Map<String, Object> buildTriggerVariablesForConditions( + Event event, Collection<ExpressionCondition> conditions) { + try { + Set<String> unionKeys = new HashSet<>(); + for (ExpressionCondition c : conditions) { + if (c == null) { + continue; + } + Set<String> keys = keySetCache == null ? null : keySetCache.get(c.source()); + if (keys == null) { + return buildTriggerVariables(event); + } + unionKeys.addAll(keys); + } + return buildPartialTriggerVariables(event, unionKeys); + } catch (RuntimeException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Building trigger condition variables failed for event " + event.getId(), + e); + } + LOG.warn( + "Building trigger condition variables failed for event {}, skipping conditions", + event.getId(), + e); + return null; + } + } + + /** + * Like {@link #buildTriggerVariables} but only materializes the merged keys in {@code + * unionKeys}, reproducing the same output>root>input precedence and normalization depth. Falls + * back to the full build when trimming is unsafe: the {@code "attributes"} sentinel is present, + * or a flattened {@code output}/{@code input} container is stored as a JSON string (which the + * full build would parse and flatten, hiding subkeys from the per-key probe). + */ + @SuppressWarnings("unchecked") + private Map<String, Object> buildPartialTriggerVariables(Event event, Set<String> unionKeys) { + if (unionKeys.contains("attributes")) { + return buildTriggerVariables(event); + } + Object rawAttrsObj = event.getAttributes(); + if (!(rawAttrsObj instanceof Map)) { + return buildTriggerVariables(event); + } + Map<String, Object> rawAttrs = (Map<String, Object>) rawAttrsObj; + if (isJsonShaped(rawAttrs.get("output")) || isJsonShaped(rawAttrs.get("input"))) { + return buildTriggerVariables(event); + } + + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Map<String, Object> partialAttrs = new HashMap<>(); + for (String key : unionKeys) { + Object v = resolveMergedKey(rawAttrs, key); + if (v != MISSING) { + partialAttrs.put(key, v); + } + } + + activation.put("attributes", partialAttrs); + partialAttrs.forEach(activation::putIfAbsent); + activation.putIfAbsent("id", event.getId().toString()); + return activation; + } + + /** + * Resolves one merged key with output>root>input precedence, normalizing only the fished value + * at the depth {@link #buildTriggerVariables} would use (root attribute = depth 1, output/input + * subkey = depth 2). Returns {@link #MISSING} when the key is absent everywhere. + */ + private static Object resolveMergedKey(Map<String, Object> rawAttrs, String key) { Review Comment: The `input` / `output` subkey promotion and the optimized lookup in `resolveMergedKey()` should probably be tightened together. Today `resolveMergedKey()` checks `output.<key>`, then the root attribute, then `input.<key>` for every event type. That means ordinary user events with `input` or `output` maps still get their nested fields promoted in the referenced-key path. If this promotion is intended only as a convenience for framework `InputEvent` / `OutputEvent`, the optimized lookup should apply it only for those matching event types; otherwise `input` and `output` should be treated as normal user attributes. Also, the current `MISSING` + `present` logic appears to preserve `putIfAbsent`'s subtle null behavior, where a lower-priority value can override a higher-priority key whose value is null. I do not think that should be part of the condition semantics. If a higher-priority layer contains the key, it should win even when the value is null; only an absent key should fall through to the next layer. This would make the lookup easier to reason about and likely remove the need for the extra `present` bookkeeping. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,393 @@ +/* + * 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.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelRuntime; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +public class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Sentinel for "no value present" (distinct from a legitimate null attribute value). */ + private static final Object MISSING = new Object(); + + /** Frozen after {@link #initPrograms}; cleared by {@link #close}. */ + @Nullable private Map<String, CelRuntime.Program> programCache; + + /** Per-source referenced merged-key sets; frozen after {@link #initPrograms}. */ + @Nullable private Map<String, Set<String>> keySetCache; + + private final ConditionEvaluationFailureStrategy failureStrategy; + + public ConditionEvaluator() { + this(ConditionEvaluationFailureStrategy.WARN_AND_SKIP); + } + + public ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Pre-compiles {@code expressions} and freezes the caches. Nulls are skipped. */ + public void initPrograms(Collection<ExpressionCondition> expressions) { + Map<String, CelRuntime.Program> programs = new HashMap<>(); + Map<String, Set<String>> keySets = new HashMap<>(); + for (ExpressionCondition expression : expressions) { + if (expression == null) { + continue; + } + String source = expression.source(); + programs.computeIfAbsent(source, ConditionExpressionCompiler::toProgram); + keySets.computeIfAbsent(source, ConditionExpressionCompiler::extractMergedKeys); + } + this.programCache = Collections.unmodifiableMap(programs); + this.keySetCache = Collections.unmodifiableMap(keySets); + } + + public void close() { + programCache = null; + keySetCache = null; + } + + /** Evaluates {@code expression} (which must have been pre-compiled). Null returns true. */ + public boolean evaluate( + @Nullable ExpressionCondition expression, Map<String, Object> activation) { + if (expression == null) { + return true; + } + String source = expression.source(); + try { + CelRuntime.Program program = programCache.get(source); + if (program == null) { + throw new IllegalStateException( + "Trigger condition was not pre-compiled via initPrograms(): \"" + + source + + "\""); + } + return evaluateProgram(source, program, activation); + } catch (CelEvaluationException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Trigger condition evaluation failed for '" + source + "'", e); + } + LOG.warn("Trigger condition evaluation failed for '{}', skipping action", source, e); + return false; + } + } + + private boolean evaluateProgram( + String condition, CelRuntime.Program program, Map<String, Object> activation) + throws CelEvaluationException { + Object result = program.eval(activation); + if (result instanceof Boolean) { + return (Boolean) result; + } + String msg = + String.format( + "Trigger condition '%s' returned non-boolean type %s, treating as false", + condition, result == null ? "null" : result.getClass().getName()); + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(msg); + } + LOG.warn(msg); + return false; + } + + /** + * Builds the variable map used to evaluate a trigger condition for one event. It always + * contains: + * + * <ul> + * <li>{@code type}: the event routing type, from {@link Event#getType()}. + * <li>{@code id}: the user-provided {@code id} attribute when present; otherwise the event + * UUID. + * <li>{@code attributes}: the normalized event attributes. + * <li>{@code EventType}: built-in event type constants, e.g. {@code EventType.InputEvent}. + * </ul> + * + * <p>Event attributes are also exposed as top-level variables: an attribute {@code score} can + * be referenced as either {@code score} or {@code attributes.score}. + * + * <p>Immediate keys of nested {@code input}/{@code output} maps are exposed as top-level + * attributes too; on collision, {@code output} keys win over root attributes, which win over + * {@code input} keys. + * + * <p>String values that look like JSON objects or arrays are parsed before evaluation; narrow + * numeric types widen to {@code long} or {@code double}. + */ + @SuppressWarnings("unchecked") + public Map<String, Object> buildTriggerVariables(Event event) { + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Object normalizedAttrs = normalizeValue(event.getAttributes(), 0); + Map<String, Object> merged = new HashMap<>(); + if (normalizedAttrs instanceof Map) { + Map<String, Object> attrs = (Map<String, Object>) normalizedAttrs; + + // Precedence: output subkeys > root attributes > input subkeys (putIfAbsent keeps the + // earliest insertion). Root iteration includes the "input"/"output" maps themselves, + // so nested paths like input.region.width keep working. + Object outputObj = attrs.get("output"); + if (outputObj instanceof Map) { + ((Map<String, Object>) outputObj).forEach(merged::putIfAbsent); + } + attrs.forEach(merged::putIfAbsent); + Object inputObj = attrs.get("input"); + if (inputObj instanceof Map) { + ((Map<String, Object>) inputObj).forEach(merged::putIfAbsent); + } + } + + activation.put("attributes", merged); + // Promote to top level for bare-identifier access; framework keys win on collision. + merged.forEach(activation::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + activation.putIfAbsent("id", event.getId().toString()); + + return activation; + } + + /** + * Builds a trimmed activation covering the union of {@code conditions}' referenced merged keys. + * Returns {@code null} under {@link ConditionEvaluationFailureStrategy#WARN_AND_SKIP} when the + * build fails, so the event's trigger conditions are skipped; rethrows as {@link + * IllegalStateException} under {@link ConditionEvaluationFailureStrategy#FAIL}. Any condition + * whose key set was not pre-compiled forces the full (untrimmed) build for safety. + */ + @Nullable + public Map<String, Object> buildTriggerVariablesForConditions( + Event event, Collection<ExpressionCondition> conditions) { + try { + Set<String> unionKeys = new HashSet<>(); + for (ExpressionCondition c : conditions) { + if (c == null) { + continue; + } + Set<String> keys = keySetCache == null ? null : keySetCache.get(c.source()); + if (keys == null) { + return buildTriggerVariables(event); + } + unionKeys.addAll(keys); + } + return buildPartialTriggerVariables(event, unionKeys); + } catch (RuntimeException e) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException( + "Building trigger condition variables failed for event " + event.getId(), + e); + } + LOG.warn( + "Building trigger condition variables failed for event {}, skipping conditions", + event.getId(), + e); + return null; + } + } + + /** + * Like {@link #buildTriggerVariables} but only materializes the merged keys in {@code + * unionKeys}, reproducing the same output>root>input precedence and normalization depth. Falls + * back to the full build when trimming is unsafe: the {@code "attributes"} sentinel is present, + * or a flattened {@code output}/{@code input} container is stored as a JSON string (which the + * full build would parse and flatten, hiding subkeys from the per-key probe). + */ + @SuppressWarnings("unchecked") + private Map<String, Object> buildPartialTriggerVariables(Event event, Set<String> unionKeys) { + if (unionKeys.contains("attributes")) { + return buildTriggerVariables(event); + } + Object rawAttrsObj = event.getAttributes(); + if (!(rawAttrsObj instanceof Map)) { + return buildTriggerVariables(event); + } + Map<String, Object> rawAttrs = (Map<String, Object>) rawAttrsObj; + if (isJsonShaped(rawAttrs.get("output")) || isJsonShaped(rawAttrs.get("input"))) { + return buildTriggerVariables(event); + } + + Map<String, Object> activation = new HashMap<>(); + activation.put("type", event.getType()); + activation.put("EventType", ConditionExpressionCompiler.EVENT_TYPE_CONSTANTS); + + Map<String, Object> partialAttrs = new HashMap<>(); + for (String key : unionKeys) { + Object v = resolveMergedKey(rawAttrs, key); + if (v != MISSING) { + partialAttrs.put(key, v); + } + } + + activation.put("attributes", partialAttrs); + partialAttrs.forEach(activation::putIfAbsent); + activation.putIfAbsent("id", event.getId().toString()); + return activation; + } + + /** + * Resolves one merged key with output>root>input precedence, normalizing only the fished value + * at the depth {@link #buildTriggerVariables} would use (root attribute = depth 1, output/input + * subkey = depth 2). Returns {@link #MISSING} when the key is absent everywhere. + */ + private static Object resolveMergedKey(Map<String, Object> rawAttrs, String key) { + // First non-null value in output>root>input order wins, mirroring the full build where + // Map.putIfAbsent lets a lower-precedence non-null overwrite a higher-precedence null. When + // the key is present only as null everywhere, resolve to null (present); MISSING otherwise. + boolean present = false; + Object out = subValue(rawAttrs.get("output"), key); + if (out != MISSING) { + present = true; + if (out != null) { + return out; + } + } + if (rawAttrs.containsKey(key)) { + present = true; + Object root = normalizeValue(rawAttrs.get(key), 1); + if (root != null) { + return root; + } + } + Object in = subValue(rawAttrs.get("input"), key); + if (in != MISSING) { + present = true; + if (in != null) { + return in; + } + } + return present ? null : MISSING; Review Comment: Can we make this helper put into `partialAttrs` directly when the key exists, instead of returning an `Object` plus using a `MISSING` sentinel? For example, `putMergedKeyIfPresent(...)` could use `containsKey()` on the relevant map and call `partialAttrs.put(key, normalizedValue)` directly. This would preserve present `null` values naturally, skip absent keys, and remove both the `MISSING` sentinel and the extra `present` bookkeeping. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; + } + conditionEvaluator = + new ConditionEvaluator( + agentPlan + .getConfig() + .get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY)); + conditionEvaluator.initPrograms(expressionConditions); + } + + /** Returns actions to fire for {@code event}: typed hits first, then CEL hits. */ + public List<Action> route(Event event) { + List<Action> typedHits = + agentPlan + .getActionsByEvent() + .getOrDefault(event.getType(), Collections.emptyList()); + + // CEL candidates = actions with at least one CEL entry, excluding those already + // matched by typed routing for this event type. This avoids double-firing. + List<Action> celCandidates; + List<Action> withCel = agentPlan.getActionsWithExpressions(); + if (withCel.isEmpty()) { + celCandidates = Collections.emptyList(); + } else { + celCandidates = new ArrayList<>(); + for (Action a : withCel) { + if (!typedHits.contains(a)) { + celCandidates.add(a); + } + } + } + + if (celCandidates.isEmpty()) { + return typedHits; + } Review Comment: ``` List<Action> expressionCandidates = new ArrayList<>(); for (Action action : agentPlan.getActionsWithExpressions()) { if (!typedHits.contains(action)) { expressionCandidates.add(action); } } if (expressionCandidates.isEmpty()) { return typedHits; } ``` ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; + } + conditionEvaluator = + new ConditionEvaluator( + agentPlan + .getConfig() + .get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY)); + conditionEvaluator.initPrograms(expressionConditions); + } + + /** Returns actions to fire for {@code event}: typed hits first, then CEL hits. */ + public List<Action> route(Event event) { + List<Action> typedHits = + agentPlan + .getActionsByEvent() + .getOrDefault(event.getType(), Collections.emptyList()); + + // CEL candidates = actions with at least one CEL entry, excluding those already + // matched by typed routing for this event type. This avoids double-firing. + List<Action> celCandidates; + List<Action> withCel = agentPlan.getActionsWithExpressions(); + if (withCel.isEmpty()) { + celCandidates = Collections.emptyList(); + } else { + celCandidates = new ArrayList<>(); + for (Action a : withCel) { + if (!typedHits.contains(a)) { + celCandidates.add(a); + } + } + } + + if (celCandidates.isEmpty()) { + return typedHits; + } + + // Preserves typed-first ordering and deduplicates. + LinkedHashSet<Action> matched = new LinkedHashSet<>(typedHits); + + Map<String, Object> activation = + conditionEvaluator.buildTriggerVariablesForConditions( + event, expressionConditionsOf(celCandidates)); + if (activation == null) { + return new ArrayList<>(matched); + } + for (Action a : celCandidates) { + // Within-action OR: first matching CEL expression admits the action. + for (ParsedCondition pc : a.getParsedConditions()) { + if (!(pc instanceof ExpressionCondition)) { + continue; + } + if (conditionEvaluator.evaluate((ExpressionCondition) pc, activation)) { + matched.add(a); + break; + } + } + } + return new ArrayList<>(matched); + } + + private static List<ExpressionCondition> expressionConditionsOf(Collection<Action> actions) { Review Comment: Suggest `extractExpressionConditions` ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.runtime.condition; + +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.plan.AgentPlan; +import org.apache.flink.agents.plan.actions.Action; +import org.apache.flink.agents.plan.condition.ParsedCondition; +import org.apache.flink.agents.plan.condition.ParsedCondition.ExpressionCondition; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; + +/** + * Routes an event to matching actions: type-index fast path first, then CEL slow path. + * + * <p>Each action fires at most once per event; typed hits ordered before CEL hits. + */ +public final class ActionRouter { + + private final AgentPlan agentPlan; + + /** Null when the plan contains no CEL expressions. */ + private ConditionEvaluator conditionEvaluator; + + public ActionRouter(AgentPlan agentPlan) { + if (agentPlan == null) { + throw new IllegalArgumentException("ActionRouter: agentPlan must not be null"); + } + this.agentPlan = agentPlan; + } + + /** Pre-compiles all CEL expressions in the plan. */ + public void open() { + List<ExpressionCondition> expressionConditions = + expressionConditionsOf(agentPlan.getActions().values()); + if (expressionConditions.isEmpty()) { + return; + } + conditionEvaluator = + new ConditionEvaluator( + agentPlan + .getConfig() + .get(AgentConfigOptions.CONDITION_EVALUATION_FAILURE_STRATEGY)); + conditionEvaluator.initPrograms(expressionConditions); + } + + /** Returns actions to fire for {@code event}: typed hits first, then CEL hits. */ + public List<Action> route(Event event) { + List<Action> typedHits = + agentPlan + .getActionsByEvent() + .getOrDefault(event.getType(), Collections.emptyList()); + + // CEL candidates = actions with at least one CEL entry, excluding those already + // matched by typed routing for this event type. This avoids double-firing. + List<Action> celCandidates; + List<Action> withCel = agentPlan.getActionsWithExpressions(); + if (withCel.isEmpty()) { + celCandidates = Collections.emptyList(); + } else { + celCandidates = new ArrayList<>(); + for (Action a : withCel) { + if (!typedHits.contains(a)) { + celCandidates.add(a); + } + } + } + + if (celCandidates.isEmpty()) { + return typedHits; + } + + // Preserves typed-first ordering and deduplicates. + LinkedHashSet<Action> matched = new LinkedHashSet<>(typedHits); Review Comment: Can this be a plain `ArrayList` instead of a `LinkedHashSet`? `actionsByEvent` and `actionsWithExpressions` should not contain duplicate actions, expression candidates are already filtered with `!typedHits.contains(a)`, and each candidate is added at most once because the inner loop breaks after the first matching expression. Using a set here suggests there is another duplicate path to handle, but under the current plan invariants there should not be one. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
