rosemarYuan commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3386875328
########## 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: Good catch — this was intended to be configurable, but the final wiring step was missing, so `FAIL` was only reachable from tests. The intended default is still `WARN_AND_SKIP`, since that is the safer streaming behavior: a single bad event or unexpected CEL evaluation result should not fail the whole job. That said, strict-semantics pipelines do need a fail-closed option, so `FAIL` should be reachable in production rather than remaining a test-only path. I added a follow-up commit to wire this through properly: - promoted `EvaluationFailurePolicy` to the API module as `CelEvaluationFailurePolicy`, following the existing `LoggerType` pattern; - exposed `CEL_EVALUATION_FAILURE_POLICY` in `AgentConfigOptions`, with `WARN_AND_SKIP` as the default; - updated `ActionRouter.open()` to read the policy from the plan config when constructing the evaluator. With this change, both `WARN_AND_SKIP` and `FAIL` are now selectable in production. Default behavior is unchanged for existing users — anyone not setting `celEvaluationFailurePolicy` in plan config still gets `WARN_AND_SKIP`. ########## 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: You're right — the doc claim and the code had drifted, and I did miss this during the implementation. I adopted the local fix you suggested: duplicating the two numeric caps in the plan-side parser is lighter than crossing the plan/runtime module boundary for a shared `CelOptions` singleton. With this change, a too-deep or too-long expression now fails at `classify()` during job submission, which matches the runtime facade behavior claimed in the documentation. -- 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]
