weiqingy commented on code in PR #924:
URL: https://github.com/apache/flink-agents/pull/924#discussion_r3697188385
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -369,36 +392,54 @@ private void processActionTaskForKey(Object key) throws
Exception {
key, sequenceNumber, actionTask.action,
actionTask.event);
}
- // Set up durable execution context for fine-grained recovery
- durableExecManager.setupDurableExecutionContext(
- actionTask, actionState, sequenceNumber);
-
- ActionTask.ActionTaskResult actionTaskResult =
- actionTask.invoke(
- getRuntimeContext().getUserCodeClassLoader(),
- this.pythonBridge.getPythonActionExecutor());
-
- // We remove the contexts from the map after the task is
processed. They will be added
- // back later if the action task has a generated action task,
meaning it is not
- // finished.
- contextManager.removeMemoryContext(actionTask);
- durableExecManager.removeDurableContext(actionTask);
- contextManager.removeContinuationContext(actionTask);
- contextManager.removePythonAwaitableRef(actionTask);
- durableExecManager.maybePersistTaskResult(
- key,
- sequenceNumber,
- actionTask.action,
- actionTask.event,
- actionTask.getRunnerContext(),
- actionTaskResult);
- isFinished = actionTaskResult.isFinished();
- outputEvents = actionTaskResult.getOutputEvents();
- generatedActionTaskOpt = actionTaskResult.getGeneratedActionTask();
+ notifyActionStarted(actionTask);
+ try {
+ // Set up durable execution context for fine-grained recovery
+ durableExecManager.setupDurableExecutionContext(
+ actionTask, actionState, sequenceNumber);
+
+ ActionTask.ActionTaskResult actionTaskResult =
+ actionTask.invoke(
+ getRuntimeContext().getUserCodeClassLoader(),
+ this.pythonBridge.getPythonActionExecutor());
+
+ // Drop task-local contexts after each step; continuations
transfer them back.
+ contextManager.removeMemoryContext(actionTask);
+ durableExecManager.removeDurableContext(actionTask);
+ contextManager.removeContinuationContext(actionTask);
+ contextManager.removePythonAwaitableRef(actionTask);
+ durableExecManager.maybePersistTaskResult(
+ key,
+ sequenceNumber,
+ actionTask.action,
+ actionTask.event,
+ actionTask.getRunnerContext(),
+ actionTaskResult);
+ isFinished = actionTaskResult.isFinished();
+ outputEvents = actionTaskResult.getOutputEvents();
+ generatedActionTaskOpt =
actionTaskResult.getGeneratedActionTask();
+ notifyFinished = isFinished;
+ } catch (Exception e) {
Review Comment:
Could the Action lifecycle guarantee cover the remaining failure paths here?
This `catch (Exception)` misses a raw `Error`, including one now unwrapped
by `JavaFunction.java:114`. Before this PR that body was just `return
getMethod().invoke(null, args);`, so every user throwable arrived wrapped in
`InvocationTargetException`. A tool throwing `AssertionError` used to be caught
at `ToolCallAction.java:174` and reported via `ExecutionReporters.failed` at
`:179`. Now it skips this catch too and fails the task at `:306`'s `catch
(Throwable t)`, leaving `_execution_started_event` with no terminal Event.
Separately, `processEvent(...)` at `:434` can throw after
`maybePersistTaskResult` at `:411` but before `notifyActionFinished` at `:437`.
This catch has already closed at `:429` and the `finally` at `:439` only calls
`completeActionExecution`, so that produces the same incomplete lifecycle, with
the result persisted.
Would catching `Throwable` long enough to report and clean up before
rethrowing, or emitting finished right after successful invocation and
persistence, keep every started Action terminal without changing which failures
stop the task?
##########
python/flink_agents/runtime/flink_runner_context.py:
##########
@@ -56,6 +59,24 @@
logger = logging.getLogger(__name__)
+def _error_type(error: BaseException) -> str:
+ return f"{error.__class__.__module__}.{error.__class__.__qualname__}"
+
+
+def _root_cause(error: BaseException) -> BaseException:
+ current = error
+ visited: set[int] = set()
+ while id(current) not in visited:
+ visited.add(id(current))
+ cause = current.__cause__
+ if cause is None and not current.__suppress_context__:
+ cause = current.__context__
Review Comment:
`_root_cause` follows `__cause__` then `__context__` when
`__suppress_context__` is false (`:71-73`), while Java walks only `getCause()`
(`ExecutionLifecycleEvents.java:100-107`). `__context__` is set implicitly by
any `raise` inside an `except` block, where `getCause()` is set only when a
cause is passed explicitly.
So a `MyError` raised inside `except JSONDecodeError` records `errorType:
json.decoder.JSONDecodeError` where Java records `MyError`. Same wrapped
failure, two `errorType` values in one log file, and a cross-language query on
that field splits. `AGENTS.md` asks that "Public API changes must keep Java,
Python, and YAML APIs semantically aligned".
Is following `__context__` deliberate? Restricting to `__cause__` would
match Java, though it may be buying something I can't see. Either way
`test_failed_execution_reports_deepest_cause` wires only `__cause__`
(`test_flink_runner_context_trace.py:39`), so that branch is unexercised on
both sides.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/EventLogWriter.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.eventlog;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventContext;
+import org.apache.flink.agents.api.logger.EventLogger;
+import org.apache.flink.agents.api.logger.EventLoggerConfig;
+import org.apache.flink.agents.api.logger.EventLoggerFactory;
+import org.apache.flink.agents.api.logger.EventLoggerOpenParams;
+import org.apache.flink.agents.api.logger.LoggerType;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.runtime.metrics.BuiltInMetrics;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import static
org.apache.flink.agents.api.configuration.AgentConfigOptions.BASE_LOG_DIR;
+import static
org.apache.flink.agents.api.configuration.AgentConfigOptions.EVENT_LOGGER_TYPE;
+import static
org.apache.flink.agents.api.configuration.AgentConfigOptions.EVENT_LOG_TRACE_ENABLED;
+
+/** Operator-scoped writer that owns the physical Event Log logger lifecycle.
*/
+@Internal
+public final class EventLogWriter implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EventLogWriter.class);
+
+ @Nullable private final EventLogger eventLogger;
+ private final boolean traceEnabled;
+
+ public static EventLogWriter create(AgentPlan agentPlan) {
+ return new EventLogWriter(
+ createEventLogger(agentPlan),
agentPlan.getConfig().get(EVENT_LOG_TRACE_ENABLED));
+ }
+
+ @VisibleForTesting
+ public static EventLogWriter forEventLogger(@Nullable EventLogger
eventLogger) {
+ return forEventLogger(eventLogger, true);
+ }
+
+ @VisibleForTesting
+ public static EventLogWriter forEventLogger(
+ @Nullable EventLogger eventLogger, boolean traceEnabled) {
+ return new EventLogWriter(eventLogger, traceEnabled);
+ }
+
+ private EventLogWriter(@Nullable EventLogger eventLogger, boolean
traceEnabled) {
+ this.eventLogger = eventLogger;
+ this.traceEnabled = traceEnabled;
+ }
+
+ public void open(StreamingRuntimeContext runtimeContext, BuiltInMetrics
builtInMetrics)
+ throws Exception {
+ if (eventLogger == null) {
+ return;
+ }
+ eventLogger.open(new EventLoggerOpenParams(runtimeContext));
+ if (eventLogger instanceof FileEventLogger) {
+ ((FileEventLogger) eventLogger)
+
.setTruncatedEventsCounter(builtInMetrics.getEventLogTruncatedEventsCounter());
+ } else if (eventLogger instanceof Slf4jEventLogger) {
+ ((Slf4jEventLogger) eventLogger)
+
.setTruncatedEventsCounter(builtInMetrics.getEventLogTruncatedEventsCounter());
+ }
+ }
+
+ /** Appends and flushes a business Event best-effort. */
+ public void appendBusinessEventAndFlush(
+ EventContext eventContext, Event event, @Nullable
ExecutionTraceContext traceContext) {
+ appendAndFlush(eventContext, event, traceEnabled ? traceContext :
null);
+ }
+
+ /** Appends and flushes an execution lifecycle Event when Trace recording
is enabled. */
+ public void appendExecutionEventAndFlush(Event event,
ExecutionTraceContext traceContext) {
+ if (!traceEnabled) {
+ return;
+ }
+ appendAndFlush(new EventContext(event), event, traceContext);
+ }
+
+ private void appendAndFlush(
+ EventContext eventContext, Event event, @Nullable
ExecutionTraceContext traceContext) {
+ if (eventLogger == null) {
+ return;
+ }
+ try {
+ eventLogger.append(eventContext, event, traceContext);
+ eventLogger.flush();
+ } catch (Exception logError) {
Review Comment:
Best-effort writes look intentional, but this also changes what an `append`
or `flush` failure does. At the merge base (`6f020c50`) the work sat in
`EventRouter.notifyEventProcessed` (`EventRouter.java:233-242`), which had no
try/catch and declared `throws Exception`, so a failure propagated and failed
the task. Was making it non-fatal the intent, or a side effect of the move?
Either way, what would you want an operator to see when a write is dropped?
`BuiltInMetrics` declares and registers `eventLogTruncatedEvents` (`:40`,
`:53`) but has no equivalent for failed writes, so an operator whose disk
filled gets a log that just stops while the job stays green. A first-failure
WARN plus an `eventLogWriteFailures` counter next to the truncation one is the
shape I had in mind, though you may be weighing log noise against it.
Smaller thing in the same block: `flush()` at `:108` is skipped when
`append` at `:107` throws, so a partial line can sit in the `PrintWriter`
buffer.
##########
docs/content/docs/operations/monitoring.md:
##########
@@ -240,7 +251,7 @@ Example record at `STANDARD` with a long string and a large
array truncated:
### Per-event-type log levels
-You can override the level for individual event types using the
`event-log.type.<EVENT_TYPE>.level` config key, where `<EVENT_TYPE>` is the
event's routing type string (the same string that appears as `eventType` in the
JSON log). Built-in events use short snake-cased names such as:
+You can override the level for individual event types using the
`event-log.type.<EVENT_TYPE>.level` config key, where `<EVENT_TYPE>` is the
event's routing type string (the same string that appears as `eventType` in the
JSON log). Although the field name uses camelCase, built-in Event type values
remain snake-cased:
Review Comment:
Would it help to list the four `_execution_*` routing types in the
per-event-type table, with a note on how they compose with
`event-log.trace.enabled`?
Since this line says the per-type key is the event's routing type string,
Trace can be enabled while `event-log.type._execution_started_event.level: OFF`
still removes the started Events. That interaction isn't visible from the table
today, and the four lifecycle types aren't listed in it at all.
##########
docs/content/docs/operations/monitoring.md:
##########
@@ -291,4 +302,6 @@ Other per-type levels from `config.yaml` are preserved —
the `-D` flag only ov
### Compatibility Notes
Review Comment:
nit: could the Compatibility Notes name the actual rewrites, `event`
removed, `event.id` → `eventId`, `event.attributes` → `eventAttributes`, and
mention that Python Event IDs changed from content-derived to per-occurrence
UUIDs?
`eventType` was already top-level, so the current wording names the one
field that didn't move, and a reader can't derive the rest. Grepping `docs/`
for content-hash or `uuid4` returns nothing, so the Python change lives only in
`event.py`, and anything relying on content-hash id equality or dedup behaves
differently after upgrade.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/eventlog/FileEventLogger.java:
##########
@@ -199,24 +207,18 @@ public void append(EventContext context, Event event)
throws Exception {
}
ObjectNode rootNode = (ObjectNode) tree;
- // Truncate the event subtree at STANDARD level.
+ // Truncate event attributes at STANDARD level.
if (level == EventLogLevel.STANDARD && truncator != null) {
- JsonNode eventNode = rootNode.get("event");
- if (eventNode instanceof ObjectNode) {
- boolean truncated = truncator.truncate((ObjectNode) eventNode);
+ JsonNode attributesNode = rootNode.get("eventAttributes");
Review Comment:
Protected names are settled, thanks. The test side is the part I'm still
unsure about.
The guarantee now lives in the two call sites (`FileEventLogger.java:212`,
`Slf4jEventLogger.java:179`) rather than in `JsonTruncator`, and I couldn't
find a test that pins it. Flipping `rootNode.get("eventAttributes")` back to
`rootNode` would fail none of the six candidate tests: the logger tests assert
only `eventAttributes.customData`, the Python e2e one only that
`"truncatedString"` appears somewhere in the line, and the `JsonTruncatorTest`
units never see a record. `eventId` is a 36-char UUID that would be wrapped at
`max-string-length=10` and nothing checks it.
That leaves the promise at `monitoring.md:219` ("Truncation only applies to
large nested content under `eventAttributes`") resting on review rather than
CI. Is one assertion in `FileEventLoggerTest.testStandardLevelTruncation`
enough to close it, checking `eventId` is still textual at
`max-string-length=10`?
--
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]
