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:
After discussion, we prefer to remain before policy, see
https://github.com/apache/flink/pull/24740#discussion_r1590908901
`setup` is just for setting some contexts.
--
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
