wenjin272 commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3628848207
########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,189 @@ +/* + * 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()); + + Map<String, Object> attrs = event.getAttributes(); + Map<String, Object> normalizedAttributes = new HashMap<>(); + for (String key : referencedTopLevelAttributeKeys) { + if (attrs.containsKey(key)) { + normalizedAttributes.put(key, normalizeValue(attrs.get(key))); + } + } + + conditionVariables.put("attributes", normalizedAttributes); + // Promote to top level for bare-identifier access; framework keys win on collision. + normalizedAttributes.forEach(conditionVariables::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + if (event.getId() != null) { + conditionVariables.putIfAbsent("id", event.getId().toString()); Review Comment: The current `id` precedence contradicts #726, which reserves root `id` and `type` for framework-owned event fields. Bare `id` should always resolve to `Event.id`, while a user attribute remains accessible as `attributes.id`. The compiler, activation construction, and `user_id_attribute_overrides_event_uuid` test should be updated accordingly. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,189 @@ +/* + * 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()); + + Map<String, Object> attrs = event.getAttributes(); + Map<String, Object> normalizedAttributes = new HashMap<>(); + for (String key : referencedTopLevelAttributeKeys) { + if (attrs.containsKey(key)) { + normalizedAttributes.put(key, normalizeValue(attrs.get(key))); + } + } + + conditionVariables.put("attributes", normalizedAttributes); + // Promote to top level for bare-identifier access; framework keys win on collision. + normalizedAttributes.forEach(conditionVariables::putIfAbsent); + // Event UUID only as fallback — a user-supplied id attribute takes precedence. + if (event.getId() != null) { + conditionVariables.putIfAbsent("id", event.getId().toString()); Review Comment: The current `id` precedence contradicts #726, which reserves root `id` and `type` for framework-owned event fields. Bare `id` should always resolve to `Event.id`, while a user attribute remains accessible as `attributes.id`. The compiler, activation construction, and `user_id_attribute_overrides_event_uuid` test should be updated accordingly. ########## e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/TriggerConditionIntegrationTest.java: ########## @@ -0,0 +1,259 @@ +/* + * 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.integration.test; + +import org.apache.flink.agents.api.AgentsExecutionEnvironment; +import org.apache.flink.agents.api.agents.Agent; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.ConditionInput; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.DoubleQuotedDottedMapKeyAgent; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.NestedMapAndAgent; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.PojoMapParityAgent; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.ScalarListPayloadAgent; +import org.apache.flink.agents.integration.test.TriggerConditionIntegrationAgent.SingleQuotedDottedMapKeyAgent; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.util.CloseableIterator; +import org.junit.jupiter.api.Test; + +import java.net.URL; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import static org.assertj.core.api.Assertions.assertThat; + +/** End-to-end test for trigger conditions in a full Flink pipeline. */ +public class TriggerConditionIntegrationTest { Review Comment: Some test cases are failing in the Java IT for Flink 1.20. ########## 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: Why was the scope changed to 'test'? This is currently causing cross-language CI failures: `NoClassDefFoundError: org/apache/logging/log4j/core/Layout` -- 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]
