wenjin272 commented on code in PR #821:
URL: https://github.com/apache/flink-agents/pull/821#discussion_r3607895559
##########
docs/content/docs/development/yaml.md:
##########
@@ -278,25 +278,42 @@ Inline action (map) fields:
|-------|----------|-------------|
| `name` | yes | Action name (unique within the agent). |
| `function` | yes | Fully-qualified callable in the form
`<module-or-class>:<qualname>`. See [Function
references](#function-references). |
-| `trigger_conditions` | yes | List of event types the action listens to.
Built-in [event aliases](#event-aliases) (`input`, `chat_request`, ...) or your
own event-type strings. |
+| `trigger_conditions` | yes | Non-empty list of exact event types or CEL
Boolean expressions. Built-in [event aliases](#event-aliases) (`input`,
`chat_request`, ...) may be used as complete entries. All entries use OR
semantics. |
Review Comment:
Do not mention CEL here.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/Action.java:
##########
@@ -23,27 +23,28 @@
import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.api.context.RunnerContext;
import org.apache.flink.agents.plan.Function;
+import org.apache.flink.agents.plan.condition.ActionSelector;
+import
org.apache.flink.agents.plan.condition.ActionSelector.ConditionExpression;
+import org.apache.flink.agents.plan.condition.ConditionExpressionValidator;
import org.apache.flink.agents.plan.serializer.ActionJsonDeserializer;
import org.apache.flink.agents.plan.serializer.ActionJsonSerializer;
import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-/**
- * Representation of an agent action with unified trigger conditions.
- *
- * <p>Each entry of {@code triggerConditions} is an event type name string.
Multiple entries combine
- * with OR.
- */
+/** Representation of an agent action with raw trigger conditions and
Plan-derived selectors. */
@JsonSerialize(using = ActionJsonSerializer.class)
@JsonDeserialize(using = ActionJsonDeserializer.class)
public class Action {
private final String name;
private final Function exec;
- private final List<String> triggerConditions;
+ private final ArrayList<String> triggerConditions;
Review Comment:
Could we keep the field typed as `List<String>` and store the unmodifiable
view once?
The defensive copy and immutability are necessary because `selectors` is
derived from these entries, and later mutations would make them inconsistent.
However, declaring the field as `ArrayList<String>` is unnecessary; `new
ArrayList<>(triggerConditions)` already guarantees the backing implementation.
```java
private final List<String> triggerConditions;
// In the constructor:
this.triggerConditions =
Collections.unmodifiableList(new ArrayList<>(triggerConditions));
// In the getter:
return triggerConditions;
```
##########
e2e-test/flink-agents-end-to-end-tests-integration/src/test/java/org/apache/flink/agents/integration/test/ConditionTriggerIntegrationAgent.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.Event;
+import org.apache.flink.agents.api.EventType;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.annotation.Action;
+import org.apache.flink.agents.api.context.RunnerContext;
+
+/**
+ * Agent exercising unified {@code @Action} selectors end-to-end: type-only,
condition-only,
+ * compound conditions, custom types, and terminal-output behavior.
+ */
+public class ConditionTriggerIntegrationAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void onAnyInput(Event event, RunnerContext ctx) {
+ Object input = InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent("all:" + input));
+ }
+
+ @Action("type == EventType.InputEvent && input > 5")
+ public static void onHighInput(Event event, RunnerContext ctx) {
+ Object input = InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent("high:" + input));
+ }
+
+ @Action("type == '_input_event' && has(input) && input > 5")
+ public static void onGuardedHighInput(Event event, RunnerContext ctx) {
+ Object input = InputEvent.fromEvent(event).getInput();
+ ctx.sendEvent(new OutputEvent("guard:" + input));
+ }
+
+ /** Re-emits every input as a custom dotted/hyphenated event type. */
+ @Action(EventType.InputEvent)
+ public static void reEmitAsCustomType(Event event, RunnerContext ctx) {
+ Object input = InputEvent.fromEvent(event).getInput();
+ Event custom = new Event("com.example.order-scored");
+ custom.setAttr("v", input);
+ ctx.sendEvent(custom);
+ }
+
+ /** Dotted/hyphenated custom type routed via the type index. */
+ @Action("'com.example.order-scored'")
+ public static void onQuotedCustomType(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent("quoted:" + event.getAttr("v")));
+ }
+
+ /** Multiple allowed types and the predicate are combined inside one CEL
expression. */
+ @Action("(type == EventType.InputEvent || type == 'never-emitted') &&
input > 5")
+ public static void onHighInputFromEitherType(Event event, RunnerContext
ctx) {
+ ctx.sendEvent(new OutputEvent("multi:" +
InputEvent.fromEvent(event).getInput()));
+ }
+
+ /** CEL short-circuiting avoids the missing attribute for every emitted
event type. */
+ @Action("type == 'never-emitted' && attributes.missing > 0")
+ public static void onNeverEmitted(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent("unexpected-non-target-evaluation"));
+ }
+
+ /** OutputEvent is a valid selector value but terminal outputs bypass
action dispatch. */
+ @Action(EventType.OutputEvent)
+ public static void onTerminalOutput(Event event, RunnerContext ctx) {
+ ctx.sendEvent(new OutputEvent("unexpected-output-routing"));
+ }
+}
Review Comment:
The E2E should cover the production paths introduced by trigger-condition
evaluation, rather than repeating the scalar `input > 5` case. Concrete
scenarios could look like the
following.
First, use a real Java POJO as the Flink input so the test exercises
`EventRouter.wrapToInputEvent()`:
```java
public static final class ConditionInput implements Serializable {
public String id;
public String status;
public int value;
public ConditionInput() {}
public ConditionInput(String id, String status, int value) {
this.id = id;
this.status = status;
this.value = value;
}
}
```
Add actions covering both explicit payload access and promoted POJO fields:
```java
@Action("type == EventType.InputEvent && input.status == 'ok'")
public static void onNestedPojoField(Event event, RunnerContext ctx) {
ConditionInput input = (ConditionInput)
InputEvent.fromEvent(event).getInput();
ctx.sendEvent(new OutputEvent("nested:" + input.id));
}
@Action("type == EventType.InputEvent && status == 'ok'")
public static void onPromotedPojoField(Event event, RunnerContext ctx) {
ConditionInput input = (ConditionInput)
InputEvent.fromEvent(event).getInput();
ctx.sendEvent(new OutputEvent("promoted:" + input.id));
}
```
This directly detects the current behavior where `input.status` works for
a Java POJO but `status` only starts working after the payload becomes a Map.
Multiple trigger entries should also be tested for OR semantics and single
execution:
```java
@Action({"status == 'ok'", "value > 5"})
public static void onStatusOrValue(Event event, RunnerContext ctx) {
ConditionInput input = (ConditionInput)
InputEvent.fromEvent(event).getInput();
ctx.sendEvent(new OutputEvent("or:" + input.id));
}
```
With these inputs:
```java
DataStream<ConditionInput> inputs =
env.fromElements(
new ConditionInput("ok-high", "ok", 9), // both
entries match
new ConditionInput("error-high", "error", 9), // value
matches
new ConditionInput("ok-low", "ok", 2)); // status
matches
```
the `or:` action should produce exactly three outputs, proving that the
action executes once when both entries match.
A custom event should exercise structured attributes and guarded missing
paths through the full action-routing path:
```java
@Action(EventType.InputEvent)
public static void emitCustomEvent(Event event, RunnerContext ctx) {
ConditionInput input = (ConditionInput)
InputEvent.fromEvent(event).getInput();
Map<String, Object> result = new HashMap<>();
result.put("status", input.status);
result.put("value", input.value);
if ("ok".equals(input.status)) {
result.put("details", Map.of("code", "ready"));
}
ctx.sendEvent(new Event("condition.result", Map.of("result", result)));
}
@Action("type == 'condition.result' && result.status == 'ok'")
public static void onCustomAttribute(Event event, RunnerContext ctx) {
ctx.sendEvent(new OutputEvent("custom"));
}
@Action(
"type == 'condition.result' "
+ "&& has(result.details.code) "
+ "&& result.details.code == 'ready'")
public static void onGuardedNestedAttribute(Event event, RunnerContext
ctx) {
ctx.sendEvent(new OutputEvent("guarded"));
}
```
Finally, add a runtime parity test for the Java and cross-language event
representations:
```java
ConditionInput payload = new ConditionInput("id", "ok", 9);
InputEvent typed = new InputEvent(payload);
Event jsonShaped =
Event.fromJson(OBJECT_MAPPER.writeValueAsString(typed));
Action action = action("status-ok", "status == 'ok'");
ActionMatcher matcher = new ActionMatcher(plan(action));
assertThat(matcher.match(typed)).containsExactly(action);
assertThat(matcher.match(jsonShaped)).containsExactly(action);
```
This test fails with the current pre-normalization `instanceof Map` check
and verifies that a direct Java POJO and the equivalent
cross-language/base-Event JSON shape have
identical condition semantics.
The classes should also be renamed to `TriggerConditionIntegrationAgent`
and `TriggerConditionIntegrationTest`, matching the user-facing term `trigger
condition`.
##########
docs/content/docs/development/yaml.md:
##########
@@ -278,25 +278,42 @@ Inline action (map) fields:
|-------|----------|-------------|
| `name` | yes | Action name (unique within the agent). |
| `function` | yes | Fully-qualified callable in the form
`<module-or-class>:<qualname>`. See [Function
references](#function-references). |
-| `trigger_conditions` | yes | List of event types the action listens to.
Built-in [event aliases](#event-aliases) (`input`, `chat_request`, ...) or your
own event-type strings. |
+| `trigger_conditions` | yes | Non-empty list of exact event types or CEL
Boolean expressions. Built-in [event aliases](#event-aliases) (`input`,
`chat_request`, ...) may be used as complete entries. All entries use OR
semantics. |
| `type` | no | Implementation language: `python` or `java`. Defaults to
`python` (see [Selecting the implementation
language](#selecting-the-implementation-language)). |
| `config` | no | Free-form configuration map passed to the action at runtime.
|
+An unquoted exact event type is an identifier or dotted identifier such as
`ready` or
+`order.created`. Each segment starts with an ASCII letter or underscore and
continues with ASCII
+letters, digits, or underscores. `EventType.*`, `true`, `false`, and `null`
are expression spellings
+instead. Boolean field conditions must be explicit, such as `ready == true` or
+`attributes.ready == true`.
+
+A quoted event type is a compatibility form for routable names such as
`'order-created'` or
Review Comment:
The quoted event-type syntax does not work as documented for YAML because
YAML removes scalar quotes during parsing.
For example:
```yaml
trigger_conditions:
- 'order-created'
```
passes `order-created` to `ActionSelector`, which is classified as an
expression rather than an `EventTypeMatch`. Users currently have to preserve
the quote characters explicitly:
```yaml
trigger_conditions:
- "'order-created'"
```
Could we document the exact YAML syntax and add a loader-to-classifier
test for it? Otherwise, a documented hyphenated event type will silently be
treated as an expression and
never route as expected.
##########
api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java:
##########
@@ -427,7 +427,7 @@ static Function resolveActionFunction(ActionSpec action) {
return resolveFunction(action.getName(), action.getFunction(),
language, paramTypes);
}
- /** Register an action on the agent with event aliases resolved. */
+ /** Register an action after resolving event aliases that occupy a
complete selector entry. */
Review Comment:
Suggested change to: `Registers an action, replacing trigger conditions that
exactly match built-in event aliases while leaving expression conditions
unchanged.`
##########
plan/src/main/java/org/apache/flink/agents/plan/condition/ActionSelector.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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 java.util.Objects;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public abstract class ActionSelector {
Review Comment:
The current naming feels inconsistent with the role of these classes.
`ActionMatcher` is the component that actually selects matching actions,
while `ActionSelector` represents one classified `trigger_conditions` entry.
The name `EventTypeMatch`
also sounds like a successful matching result, although it is actually a
matching rule. `ConditionExpression` describes the rule from a different
perspective, so the two subtype
names are not parallel.
I suggest aligning the hierarchy with the existing user-facing
`trigger_conditions` concept:
```java
abstract class TriggerCondition
final class EventTypeCondition extends TriggerCondition
final class ExpressionCondition extends TriggerCondition
```
This makes it clear that each object represents one trigger condition and
that the two subclasses are alternative condition types.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/condition/ActionMatcher.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.api.configuration.AgentConfigOptions;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.plan.actions.Action;
+import org.apache.flink.agents.plan.condition.ActionSelector;
+import
org.apache.flink.agents.plan.condition.ActionSelector.ConditionExpression;
+import org.apache.flink.agents.plan.condition.ActionSelector.EventTypeMatch;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Matches actions against event types and condition expressions. */
+public final class ActionMatcher {
+
+ private final Map<String, Set<Action>> actionsByEventType;
+ private final Map<Action,
List<ConditionExpressionCompiler.CompiledCondition>>
+ compiledConditionsByAction;
+ private final ConditionEvaluator conditionEvaluator;
+
+ public ActionMatcher(AgentPlan agentPlan) {
+ if (agentPlan == null) {
+ throw new IllegalArgumentException("ActionMatcher: agentPlan must
not be null");
+ }
+ Map<String, Set<Action>> eventTypeIndex = new LinkedHashMap<>();
+ Map<Action, List<ConditionExpressionCompiler.CompiledCondition>>
conditionIndex =
+ new LinkedHashMap<>();
+ for (Action action : agentPlan.getActions().values()) {
+ List<ConditionExpressionCompiler.CompiledCondition>
actionConditions =
+ new ArrayList<>();
+ for (ActionSelector selector : action.getSelectors()) {
+ if (selector instanceof EventTypeMatch) {
+ String eventType = ((EventTypeMatch) selector).eventType();
+ // Raw selectors preserve duplicates; the runtime index
keeps each action once.
+ eventTypeIndex
+ .computeIfAbsent(eventType, ignored -> new
LinkedHashSet<>())
+ .add(action);
+ continue;
+ }
+
+ ConditionExpression condition = (ConditionExpression) selector;
+
actionConditions.add(ConditionExpressionCompiler.compile(condition));
+ }
Review Comment:
Suggested change to:
```
for (TriggerCondition condition : action.getTriggerConditions()) {
if (condition instanceof EventTypeCondition) {
...
} else if (condition instanceof ExpressionCondition) {
...
} else {
throw new IllegalStateException(
"Unsupported condition type: " +
condition.getClass().getName());
}
}
```
##########
api/src/main/java/org/apache/flink/agents/api/yaml/spec/ActionSpec.java:
##########
@@ -44,16 +44,12 @@ public final class ActionSpec {
public ActionSpec(
@JsonProperty(value = "name", required = true) String name,
@JsonProperty("function") String function,
- @JsonProperty(value = "trigger_conditions", required = true)
- List<String> triggerConditions,
+ @JsonProperty("trigger_conditions") List<String> triggerConditions,
@JsonProperty("config") Map<String, Object> config,
@JsonProperty("type") Language type) {
- if (triggerConditions == null || triggerConditions.isEmpty()) {
- throw new IllegalArgumentException("trigger_conditions must not be
empty");
- }
this.name = name;
this.function = function;
- this.triggerConditions = triggerConditions;
+ this.triggerConditions = triggerConditions == null ? List.of() :
triggerConditions;
Review Comment:
Why is `trigger_conditions` changed from a required, non-empty field to a
nullable field here?
Field presence and non-emptiness are YAML structural constraints. Selector
classification and expression validation can be deferred to the Plan layer
without allowing an invalid
`ActionSpec`.
The current change lets invalid YAML pass `ActionSpec` and
`YamlLoader.buildAgents()`, delays the error until `AgentPlan` construction,
and creates an unnecessary Java/Python
schema-parity exception because Python still requires `Field(...,
min_length=1)`.
Please keep `trigger_conditions` required and reject `null` or empty lists
in `ActionSpec`, while leaving per-entry classification and expression
validation to Plan.
##########
plan/src/main/java/org/apache/flink/agents/plan/condition/ConditionSyntaxPolicy.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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 com.google.common.collect.ImmutableList;
+import dev.cel.common.CelOptions;
+import dev.cel.common.ast.CelExpr;
+import dev.cel.parser.CelMacro;
+import dev.cel.parser.CelMacroExprFactory;
+import dev.cel.parser.CelParser;
+import dev.cel.parser.CelParserFactory;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+/** Shared syntax configuration for trigger conditions. */
+public final class ConditionSyntaxPolicy {
Review Comment:
Would `ConditionExpressionDialect` be a clearer name for this class?
`ConditionSyntaxPolicy` sounds as though the class only defines syntax
rules, while it also creates the shared parser, configures parse and evaluation
limits, implements the
custom `has()` macro expansion, and maintains the reserved identifiers
used by that expansion.
`ConditionExpressionDialect` may better reflect this broader
responsibility. The CEL-specific objects and macro implementation details, such
as `HAS` and the reserved identifier
set, could also remain private behind parsing/options methods.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+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.
+ */
+ @SuppressWarnings("unchecked")
+ Map<String, Object> buildConditionVariables(
Review Comment:
`buildConditionVariables()` should be reworked because the current
three-source merge does not match the event model and produces different
behavior for Java POJO payloads and
cross-language events.
An event belongs to exactly one category, with one valid attribute shape:
```text
InputEvent: attributes = {"input": payload}
OutputEvent: attributes = {"output": payload}
Other Event: attributes = arbitrary user attributes
```
The implementation should therefore select behavior by `event.getType()`.
It must not use Java subclass checks because events received across the
Python/Java boundary are
deserialized as base `Event` instances.
There should be no `output payload > root attributes > input payload`
precedence. An InputEvent should not have sibling user attributes alongside
`input`, and an OutputEvent
should not have sibling user attributes alongside `output`. Unexpected
sibling attributes should be rejected at the event boundary, or at least
ignored here, rather than being
assigned merge semantics.
The current checks also happen too early:
```java
input instanceof Map
output instanceof Map
```
Condition values use their JSON-compatible representation. A Java POJO
stored as a normal attribute already supports nested access because
`normalizeValue()` converts it through
Jackson. The same rule should apply to input/output payloads.
Currently:
```java
new InputEvent(new Result("ok", 42))
```
supports:
```cel
input.status == "ok"
```
but not the promoted form:
```cel
status == "ok"
```
After a JSON or cross-language round trip, the payload becomes a Map and
the promoted form starts working. The behavior therefore depends on whether the
event crossed the language
boundary.
For built-in input/output events, the implementation should expose the
payload itself under `input`/`output` and lazily convert an object-shaped
payload when one of its fields
needs to be promoted. Other events should continue to expose their normal
attributes.
For example:
```java
String payloadKey = null;
if (InputEvent.EVENT_TYPE.equals(event.getType())) {
payloadKey = "input";
} else if (OutputEvent.EVENT_TYPE.equals(event.getType())) {
payloadKey = "output";
}
Map<String, Object> merged = new HashMap<>();
if (payloadKey == null) {
// A regular Event exposes its referenced attributes.
for (String key : referencedKeys) {
mergeAttributeIfAbsent(merged, attrs, key);
}
} else {
Object payload = attrs.get(payloadKey);
Map<String, Object> payloadAttrs = null;
for (String key : referencedKeys) {
if (payloadKey.equals(key)) {
// Preserve explicit access such as input.status or
output.status.
mergeAttributeIfAbsent(merged, attrs, key);
continue;
}
// Bare keys such as status are promoted from the only valid
payload.
if (payloadAttrs == null) {
payloadAttrs = asConditionObject(payload);
}
mergeAttributeIfAbsent(merged, payloadAttrs, key);
}
}
```
`asConditionObject()` can preserve an existing Map and otherwise normalize
a Jackson-serializable POJO before checking whether its JSON representation is
an object:
```java
@SuppressWarnings("unchecked")
private static Map<String, Object> asConditionObject(Object value) {
if (value instanceof Map) {
return (Map<String, Object>) value;
}
Object normalized = normalizeValue(value);
return normalized instanceof Map
? (Map<String, Object>) normalized
: Map.of();
}
```
This conversion remains lazy: a type-only condition does not serialize the
payload, while a condition that references `input`, `output`, or a promoted
payload field performs only
the required conversion.
Please also replace the precedence tests that construct input/output
events with sibling root attributes. The relevant tests should instead verify:
- InputEvent promotes fields only from `input`;
- OutputEvent promotes fields only from `output`;
- direct Java POJO payloads and JSON-shaped base Events behave identically;
- scalar and list payloads remain accessible as `input`/`output` but have
no promoted fields;
- type-only conditions do not serialize payloads;
- malformed built-in event envelopes are rejected or ignored consistently.
--
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]
