xintongsong commented on code in PR #103:
URL: https://github.com/apache/flink-agents/pull/103#discussion_r2268370528
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -176,13 +184,26 @@ public void open() throws Exception {
mailboxProcessor = getMailboxProcessor();
+ // Initialize the event logger if it is set.
+ initEventLogger();
+
// Since an operator restart may change the key range it manages due
to changes in
// parallelism,
// and {@link tryProcessActionTaskForKey} mails might be lost,
// it is necessary to reprocess all keys to ensure correctness.
tryResumeProcessActionTasks();
}
+ private void initEventLogger() throws Exception {
+ if (eventLogger == null) {
+ return;
+ }
+ if (eventLogger.unwrap() instanceof FlinkStateEventLogger) {
+ ((FlinkStateEventLogger)
eventLogger).setStreamingRuntimeContext(getRuntimeContext());
+ }
Review Comment:
But why not just use the original event logger when we need to accept all
event?
It's probably not a big deal, just different ways of implementation. Just
it's a bit against intuition for me, that we turn one event logger into another
event logger with a filter that does nothing.
--
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]
