weiqingy commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3386126020
########## plan/src/main/java/org/apache/flink/agents/plan/condition/ParsedCondition.java: ########## @@ -0,0 +1,140 @@ +/* + * 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.CelExpr; +import dev.cel.parser.CelParser; +import dev.cel.parser.CelParserFactory; + +import java.util.Objects; + +/** + * A parsed {@code Action.triggerConditions} entry — either {@link TypeMatch} or {@link + * CelExpression}. {@link #classify} turns a raw entry string into one of the two. + */ +public interface ParsedCondition { + + /** Original user-written entry string. */ + String source(); + + /** Parser with the custom {@code has()} macro; same dialect as the runtime facade parser. */ Review Comment: This parser is built without the resource-limit `CelOptions` that `CelExpressionFacade.CEL_PARSER` uses (`maxParseRecursionDepth(32)`, `maxExpressionCodePointSize(8192)`), so the "same dialect as the runtime facade parser" note isn't quite accurate — the parse limits differ. The practical effect is small since `classify()` runs at plan-build time on the author's own annotation strings, but it does mean a deep expression parses fine here and only trips the cap later at `toProgram`. Could we apply the same `maxParseRecursionDepth(32)` / `maxExpressionCodePointSize(8192)` caps here too? `ParsedCondition` is in `plan` and the facade's `CEL_OPTIONS` is private to `runtime`, so a shared parser would cross the module boundary — but duplicating the two caps would line the limits up with the "same dialect" doc and surface a too-deep expression once, early. ########## .github/workflows/cel-conformance.yml: ########## @@ -0,0 +1,137 @@ +# 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. + +name: CEL Cross-Language Conformance + +on: + push: + branches: [ main, 'release-*' ] + paths: + - 'runtime/src/main/java/org/apache/flink/agents/runtime/condition/**' + - 'runtime/src/test/java/org/apache/flink/agents/runtime/condition/**' + - 'e2e-test/cel-fixtures/**' + - 'plan/src/main/java/org/apache/flink/agents/plan/condition/**' + - 'python/flink_agents/runtime/condition/**' + - 'python/flink_agents/plan/condition/**' + - 'python/flink_agents/runtime/tests/test_local_runner_condition.py' + - 'python/flink_agents/runtime/tests/test_local_runner_mixed_or_dedup.py' + - '.github/workflows/cel-conformance.yml' + pull_request: + branches: [ main, 'release-*' ] + paths: + - 'runtime/src/main/java/org/apache/flink/agents/runtime/condition/**' + - 'runtime/src/test/java/org/apache/flink/agents/runtime/condition/**' + - 'e2e-test/cel-fixtures/**' + - 'plan/src/main/java/org/apache/flink/agents/plan/condition/**' + - 'python/flink_agents/runtime/condition/**' + - 'python/flink_agents/plan/condition/**' + - 'python/flink_agents/runtime/tests/test_local_runner_condition.py' + - 'python/flink_agents/runtime/tests/test_local_runner_mixed_or_dedup.py' + - '.github/workflows/cel-conformance.yml' + workflow_dispatch: + +jobs: + conformance: + name: cel-conformance + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + # Fixtures live in e2e-test/cel-fixtures/ (single source of truth). + # Java reads them via pom.xml <testResource>; Python via symlink. + # No diff step needed — both sides reference the same physical files. + + # Cross-language check: CEL reserved keyword sets must be identical. + # Skipped automatically if the Python side hasn't landed yet (PR2 alone). + - name: Ensure CEL reserved keywords are identical across languages + run: | + set -euo pipefail + if [ ! -f python/flink_agents/plan/condition/cel_reserved.py ]; then + echo "Python cel_reserved.py not present yet; skipping cross-language check." + exit 0 + fi + python3 << 'PY' + import re, sys + + java_file = 'plan/src/main/java/org/apache/flink/agents/plan/condition/CelReserved.java' Review Comment: The reserved-keyword parity step reads symbols that don't exist at head, so it verifies nothing today and will mis-fire once PR3 lands. `java_file` points at `CelReserved.java`, but the reserved set lives in `CelMacroPolicy.RESERVED_IDENTIFIERS`; the `set\.add("...")` regex matches nothing (the set is built with `Set.of(...)` + `set.addAll(CEL_STANDARD_MACROS)`), so `java_names` comes out empty; and the `DISALLOWED_MACROS` branch keys off a constant that isn't there (`CEL_STANDARD_MACROS` / `ALLOWED_MACROS` are). Line 111 also lists `AstRewriterTest` in `-Dtest`, which has no class at head. These read like leftovers from the pre-refactor design (a `CelReserved` class, a `DISALLOWED_MACROS` set, an AST rewriter) that got folded into `CelMacroPolicy` without the workflow following along. It's harmless right now because the step `exit 0`s while `cel_reserved.py` is absent — but the moment PR3 adds the Python side, this compares an empty Java set against the Python frozenset: either a spurious failure (`Java only: []`) or, if both happen to be empty, a false `OK: 0 reserved identifiers aligned` that silently passes a check guaranteeing nothing. Since the cross-language parity guard is the whole point of the workflow, would it be worth fixing it here — point the parser at `CelMacroPolicy.java`, extract `RESERVED_IDENTIFIERS` from the `Set.of(...)` block plus `CEL_STANDARD_MACROS`, drop `AstRewriterTest` — rather than carrying it forward broken? A guard that fails when `java_names` is empty would also keep a future refactor from silently re-breaking it. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionRouter.java: ########## @@ -0,0 +1,124 @@ +/* + * 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.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.CelExpression; + +import java.util.ArrayList; +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 CelConditionEvaluator 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<CelExpression> celExpressions = new ArrayList<>(); + for (Action action : agentPlan.getActions().values()) { + for (ParsedCondition pc : action.getParsedConditions()) { + if (pc instanceof CelExpression) { + celExpressions.add((CelExpression) pc); + } + } + } + if (celExpressions.isEmpty()) { + return; + } + conditionEvaluator = new CelConditionEvaluator(); Review Comment: `open()` always constructs the evaluator with the default `WARN_AND_SKIP` policy, and nothing else (config, plan field, ctor arg) can select `FAIL`, so production always silently drops an action on a CEL runtime error or non-boolean result, with only a WARN log as signal. `FAIL` is fully implemented and tested but reachable only from tests. Two things I wanted to check rather than assume: is silent-skip the intended *only* production behavior here (fail-open vs. fail-closed is a real choice — a compilable expression that throws on every event would just quietly stop firing), and is `FAIL` meant to become a config knob, or is it speculative for now? If it's not going to be selectable, dropping it would avoid the test-only code path; if it is, the wiring looks like it's missing. -- 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]
