weiqingy commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3393325778
########## .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: Thanks — the LICENSE/NOTICE additions cover it: `dev.cel:cel`/`dev.cel:protobuf` under Apache 2.0, `re2j`/`antlr4-runtime` under BSD-3-Clause, plus the guava/checker-qual/tree-sitter transitives, with the individual license files alongside. The definitive check is the shaded `dist` jar at release, but the set lines up with dev.cel's closure. And agreed on removing the workflow until the Python side lands — that's cleaner than carrying a parity check that can't run yet, and it sidesteps the empty-set false-pass. I'll keep an eye out for the rebuilt check on the Python PR: that the regex pulls a non-empty `RESERVED_IDENTIFIERS` out of `CelMacroPolicy`, and that an empty `java_names` fails loudly rather than aligning two empty sets. Quick turnaround, appreciated. ########## 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: Confirmed — the duplicated caps line up with `CelExpressionFacade.CEL_OPTIONS` (`maxParseRecursionDepth(32)` / `maxExpressionCodePointSize(8192)`), so a too-deep or too-long expression now trips at `classify()` as the doc claims. ########## 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: The wiring looks right — `getConfig()` is always initialized (every `AgentPlan` constructor sets a default `AgentConfiguration`), so `open()` won't NPE, and the enum round-trips through config the same way `EVENT_LOGGER_TYPE`/`LoggerType` does. One gap on the path you just added, though: nothing tests it end-to-end. The only `FAIL` test (`CelConditionEvaluatorTest.testFailPolicyThrowsOnEvaluationError`) constructs the evaluator directly, so it never exercises `CEL_EVALUATION_FAILURE_POLICY → ActionRouter.open() → evaluator` — and all the `ActionRouterTest` cases run on the default policy. A regression in the wiring (wrong option read, default flipped, `open()` dropping the value) would stay green. Would it be worth a small `ActionRouterTest` case — build a plan with a CEL action, `getConfig().set(AgentConfigOptions.CEL_EVALUATION_FAILURE_POLICY, FAIL)`, then assert `route()` throws `IllegalStateException` on an event whose condition errors? That would lock down the path this commit is here to enable. -- 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]
