joeyutong commented on code in PR #955:
URL: https://github.com/apache/flink-agents/pull/955#discussion_r3886227028
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -251,35 +251,41 @@ public void processElement(StreamRecord<IN> record)
throws Exception {
if (record.hasTimestamp()) {
inputEvent.setSourceTimestamp(record.getTimestamp());
}
+ builtInMetrics.markInputEventReceived(inputEvent);
- eventRouter.getKeySegmentQueue().addKeyToLastSegment(getCurrentKey());
+ Object key = getCurrentKey();
+ try {
+ eventRouter.getKeySegmentQueue().addKeyToLastSegment(key);
- if (stateManager.hasMoreActionTasks()) {
- // If there are already actions being processed for the current
key, the newly incoming
- // event should be queued and processed later. Therefore, we add
it to
- // pendingInputEventsState.
- stateManager.addPendingInputEvent(inputEvent);
- } else {
- // Otherwise, the new event is processed immediately.
- processInputEvent(getCurrentKey(), inputEvent);
+ if (stateManager.hasMoreActionTasks()) {
+ // If there are already actions being processed for the
current key, the newly
+ // incoming event should be queued and processed later.
Therefore, we add it to
+ // pendingInputEventsState.
+ enqueuePendingInputEvent(inputEvent);
+ return;
+ }
+ } catch (Exception e) {
+ builtInMetrics.markInputEventFailed(inputEvent);
+ throw e;
}
+
+ // Otherwise, the new event is processed immediately. Its failures are
attributed to the
+ // input run created by processInputEvent.
+ processInputEvent(key, inputEvent);
}
/** Resolves one context key for an input and reuses it for the entire
agent run. */
private void processInputEvent(Object key, Event inputEvent) throws
Exception {
- processEvent(key, resolveContextKey(key), inputEvent);
- }
-
- /**
- * Processes an incoming event for the given key and may submit a new mail
- * `tryProcessActionTaskForKey` to continue processing.
- */
- private void processEvent(Object key, String contextKey, Event event)
throws Exception {
- processEvent(
- key,
- contextKey,
- event,
- ExecutionTraceContext.forInputRun(contextKey,
agentPlan.getAgentName()));
+ String contextKey = resolveContextKey(key);
Review Comment:
Fixed. Context-key and trace-context setup failures now close the received
Input Event via `markInputEventFailed` before an Input Run is started. Added an
operator regression test for key-conversion failure.
--
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]
