masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590495309
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/asyncprocessing/AbstractAsyncStateStreamOperator.java:
##########
@@ -74,15 +76,20 @@ public void setup(
final int asyncBufferSize =
environment.getExecutionConfig().getAsyncStateBufferSize();
final long asyncBufferTimeout =
environment.getExecutionConfig().getAsyncStateBufferTimeout();
- // TODO: initial state executor and set state executor for aec
+
+ AsyncKeyedStateBackend asyncKeyedStateBackend =
+ Preconditions.checkNotNull(
+ stateHandler.getAsyncKeyedStateBackend(),
+ "Current State Backend doesn't support async access");
this.asyncExecutionController =
new AsyncExecutionController(
mailboxExecutor,
- null,
+ asyncKeyedStateBackend.createStateExecutor(),
maxParallelism,
asyncBufferSize,
asyncBufferTimeout,
inFlightRecordsLimit);
+ asyncKeyedStateBackend.setup(asyncExecutionController);
Review Comment:
Thanks for the suggestion.
I think it's reasonable to maintain `setup` in AEC.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
