wenjin272 commented on code in PR #821:
URL: https://github.com/apache/flink-agents/pull/821#discussion_r3687967305
##########
api/src/main/java/org/apache/flink/agents/api/yaml/spec/ActionSpec.java:
##########
@@ -58,6 +70,32 @@ public ActionSpec(
this.type = type;
}
+ private static List<String> parseTriggerConditions(
+ String actionName, JsonNode triggerConditionsNode) {
+ if (triggerConditionsNode == null || triggerConditionsNode.isNull()) {
Review Comment:
Java YAML validation should match the Python model and generated schema.
`parseTriggerConditions()` currently preserves `null` and whitespace-only
entries, while Python/schema require every entry to be a string containing at
least one non-whitespace character. This makes the same YAML pass
`YamlLoader.buildAgents()` in Java but fail immediately in Python or schema
validation.
We should reject `null` and whitespace-only entries here.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.common.values.NullValue;
+import dev.cel.runtime.CelEvaluationException;
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventType;
+import
org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Evaluates trigger condition expressions against event data. */
+final class ConditionEvaluator {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ConditionEvaluator.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private final ConditionEvaluationFailureStrategy failureStrategy;
+
+ ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) {
+ this.failureStrategy = failureStrategy;
+ }
+
+ /** Builds only the variables needed by one reached condition, then
evaluates it. */
+ boolean evaluate(ConditionExpressionCompiler.CompiledCondition condition,
Event event) {
+ Map<String, Object> conditionVariables;
+ try {
+ conditionVariables = buildConditionVariables(event, condition);
+ } catch (RuntimeException e) {
+ return handleConditionFailure(
+ "Building trigger condition variables failed for event " +
event.getId(), e);
+ }
+
+ String source = condition.source();
+ Object result;
+ try {
+ result = condition.program().eval(conditionVariables);
+ } catch (CelEvaluationException e) {
+ return handleConditionFailure(
+ "Trigger condition evaluation failed for '" + source +
"'", e);
+ }
+ if (result instanceof Boolean) {
+ return (Boolean) result;
+ }
+ return handleConditionFailure(
+ String.format(
+ "Trigger condition '%s' returned non-boolean type %s",
+ source, result == null ? "null" :
result.getClass().getName()),
+ null);
+ }
+
+ /**
+ * Applies the failure strategy to one condition failure: {@code FAIL}
throws, otherwise the
+ * failure is logged and the condition is treated as false.
+ */
+ private boolean handleConditionFailure(String message, Throwable cause) {
+ if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) {
+ throw new IllegalStateException(message, cause);
+ }
+ LOG.warn("{}, treating this condition as false", message, cause);
+ return false;
+ }
+
+ /**
+ * Builds the condition variables required for {@code event} from the
referenced keys of {@code
+ * condition}.
+ *
+ * <p>The {@code attributes} entry remains the explicit root namespace for
the event attribute
+ * map. This is needed for literal keys that cannot be represented as
identifiers, such as
+ * {@code attributes["www.andriod.com"].ip}; simple keys are also promoted
for bare-identifier
+ * access.
+ */
+ Map<String, Object> buildConditionVariables(
+ Event event, ConditionExpressionCompiler.CompiledCondition
condition) {
+ Set<String> referencedTopLevelAttributeKeys =
condition.referencedTopLevelAttributeKeys();
+
+ Map<String, Object> conditionVariables = new HashMap<>();
+ conditionVariables.put("type", event.getType());
+ conditionVariables.put("EventType", EventType.allConstants());
+ if (event.getId() != null) {
Review Comment:
Why should `ConditionEvaluator` support an event with a missing ID? Every
event reaching condition matching through the normal Java and cross-language
runtime paths has a non-null ID. This test constructs a state that is only
possible through incomplete JSON or an explicitly null ID, and unnecessarily
broadens the evaluator contract. I think the null-handling branch and this test
should be removed; any missing-ID policy belongs at the `Event` deserialization
boundary.
##########
python/flink_agents/runtime/remote_execution_environment.py:
##########
@@ -45,6 +45,8 @@
_CONFIG_FILE_NAME = "config.yaml"
_LEGACY_CONFIG_FILE_NAME = "flink-conf.yaml"
+_AGENT_PLAN_PREFLIGHT_CLASS = "org.apache.flink.agents.plan.AgentPlanPreflight"
+_AGENT_PLAN_PREFLIGHT_METHOD = "findValidationError"
Review Comment:
`AgentPlanPreflight` sounds broader than its actual responsibility. This
class does not perform general pre-execution checks; it validates
Python-produced AgentPlan JSON by deserializing it with the authoritative Java
model. A name such as `AgentPlanJsonValidator` and `validateAgentPlan` would
make that scope clearer.
##########
plan/pom.xml:
##########
@@ -89,11 +94,13 @@ under the License.
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>${log4j2.version}</version>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>${log4j2.version}</version>
+ <scope>test</scope>
Review Comment:
Changing the Log4j dependencies to `test` scope may be reasonable on its
own, but it is unrelated to CEL trigger conditions and changes the runtime
packaging contract. It has also required compensating Log4j dependencies in the
cross-language E2E module.
I think the scope changes and their E2E compensation should be reverted from
this PR and handled separately, together with the corresponding dist/NOTICE
updates and deployment verification.
##########
python/flink_agents/runtime/remote_execution_environment.py:
##########
@@ -107,10 +109,30 @@ def apply(self, agent: Agent | str) -> "AgentBuilder":
for type, name_to_resource in self.__resources.items():
agent.resources[type] = name_to_resource | agent.resources[type]
- self.__agent_plan = AgentPlan.from_agent(agent, self.__config)
+ candidate_plan = AgentPlan.from_agent(agent, self.__config)
+ candidate_plan_json =
candidate_plan.model_dump_json(serialize_as_any=True)
Review Comment:
The same `AgentPlan` is serialized in `apply()` for Java validation and
then serialized again in `to_datastream()` for `connectToAgent()`. Store the
validated JSON and pass that exact value to `connectToAgent()` instead.
This also guarantees that the submitted plan is exactly the snapshot that
passed validation, even if the shared configuration is modified after `apply()`.
Suggest `cadidate_plan` -> `agent_plan` and `candidate_plan_json` ->
`agent_plan_json`
--
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]
