You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "masteryhx (via GitHub)" <gi...@apache.org> on 2024/04/29 08:08:08 UTC
[PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
masteryhx opened a new pull request, #24740:
URL: https://github.com/apache/flink/pull/24740
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Bridge between AsyncKeyedStateBackend and AsyncExecutionController, which make aec usable with StateBackend
## Brief change log
- Bridge between AsyncKeyedStateBackend and AsyncExecutionController
- Support AsyncKeyedStateBackend for ForStKeyedStateBackend
## Verifying this change
This change added tests and can be verified as follows:
- Introduce method of building Async State Backend in StateBackendTestUtils
- Modify AsyncExecutionControllerTest, AbstractAsyncStateStreamOperatorTest, AbstractAsyncStateStreamOperatorV2Test
- Add testAsyncValueStateInstantiation in StreamingRuntimeContextTest
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590908901
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
As discussed offline, Considering minimized dependencies between `AEC` and `AsyncKeyedStateBackend`, we use before policy.
`AEC` should only know `StateExecutor` and `AsyncKeyedStateBackend` only know `StateRequestHandler`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24740:
URL: https://github.com/apache/flink/pull/24740#issuecomment-2082126122
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fd776b22665c0a416a95bcbcf0173e636434d8c1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fd776b22665c0a416a95bcbcf0173e636434d8c1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fd776b22665c0a416a95bcbcf0173e636434d8c1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590642587
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
Well, I have different opinions... I suggest keeping `AEC` and `AKSB` simple by only keeping the necessary knowledge. It seems they work well even though they don't know each other. So I'd prefer do the binding within the operator. The UT is more easy to build if the `AEC` only takes `StateExecutor` in constructor. WDTY?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java:
##########
@@ -176,15 +180,31 @@ public StreamOperatorStateContext streamOperatorStateContext(
try {
// -------------- Keyed State Backend --------------
- keyedStatedBackend =
- keyedStatedBackend(
- keySerializer,
- operatorIdentifierText,
- prioritizedOperatorSubtaskStates,
- streamTaskCloseableRegistry,
- metricGroup,
- managedMemoryFraction,
- statsCollector);
+ // TODO: Only init async keyed state backend if supported, consider to init different
+ // state backend by the operator type.
+ if (stateBackend.supportsAsyncKeyedStateBackend()) {
+ asyncKeyedStateBackend =
+ keyedStatedBackend(
+ keySerializer,
+ operatorIdentifierText,
+ prioritizedOperatorSubtaskStates,
+ streamTaskCloseableRegistry,
+ metricGroup,
+ managedMemoryFraction,
+ statsCollector,
+ StateBackend::createAsyncKeyedStateBackend);
+ } else {
Review Comment:
Do you mean only create one type of state backend (sync/async). I'd suggest splitting this up, like https://github.com/apache/flink/pull/24697/files#diff-60f97b32726f78481f7852ea82daa2f3c9919933c93c234a049442195ccf0e2b does.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590611811
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
Thanks for the clarification👍, let it stay as is for now.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590609918
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java:
##########
@@ -54,7 +59,14 @@ public class ForStKeyedStateBackendBuilder<K>
protected final Logger logger = LoggerFactory.getLogger(getClass());
+ private static final int KEY_SERIALIZER_BUFFER_START_SIZE = 32;
+
+ private static final int VALUE_SERIALIZER_BUFFER_START_SIZE = 128;
Review Comment:
It's just an empirical number which is consistent with RocksDBStateBackend.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590608598
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java:
##########
@@ -242,6 +248,25 @@ private KeyedStateStore checkPreconditionsAndGetKeyedStateStore(
return keyedStateStore;
}
+ // TODO: Reconstruct this after StateManager is ready in FLIP-410.
Review Comment:
I have renamed it.
For stateDescriptorV2 and stateDescriptorV1, I think we could only support async access for stateDescriptorV2.
It's also doable in the future to support sync access for stateDescriptorV2 which should be our inner logic.
As for supporting async access for stateDescriptorV1, it should be a bigger topic and may need a FLIP. We could just resolve this firstly.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590908392
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
As discussed offline, Considering minimized dependencies between `AEC` and `AsyncKeyedStateBackend`, we use before policy.
`AEC` should only know `StateExecutor` and `AsyncKeyedStateBackend` only know `StateRequestHandler`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx closed pull request #24740: [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController
URL: https://github.com/apache/flink/pull/24740
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "jectpro7 (via GitHub)" <gi...@apache.org>.
jectpro7 commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1584465123
##########
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:
the dependency looks like not very clear, can we make it like this?
```
new AsyncExecutionController(
mailboxExecutor,
asyncKeyedStateBackend,
maxParallelism,
asyncBufferSize,
asyncBufferTimeout,
inFlightRecordsLimit);
```
The AEC is responsible for creating the `stateExecutor` and set it up inside the construction method.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590911242
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java:
##########
@@ -176,15 +180,31 @@ public StreamOperatorStateContext streamOperatorStateContext(
try {
// -------------- Keyed State Backend --------------
- keyedStatedBackend =
- keyedStatedBackend(
- keySerializer,
- operatorIdentifierText,
- prioritizedOperatorSubtaskStates,
- streamTaskCloseableRegistry,
- metricGroup,
- managedMemoryFraction,
- statsCollector);
+ // TODO: Only init async keyed state backend if supported, consider to init different
+ // state backend by the operator type.
+ if (stateBackend.supportsAsyncKeyedStateBackend()) {
+ asyncKeyedStateBackend =
+ keyedStatedBackend(
+ keySerializer,
+ operatorIdentifierText,
+ prioritizedOperatorSubtaskStates,
+ streamTaskCloseableRegistry,
+ metricGroup,
+ managedMemoryFraction,
+ statsCollector,
+ StateBackend::createAsyncKeyedStateBackend);
+ } else {
Review Comment:
As discussed offline, Since current creation of `KeyedStateBackend` contains some allocation of resources, e.g. managed memory, we remain creating one type here.
Also add a TODO to support `KeyedStateBackend` for `AsyncKeyedStateBackend` in the future.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1591946404
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java:
##########
@@ -114,6 +116,10 @@ public void setKeyedStateStore(@Nullable KeyedStateStore keyedStateStore) {
this.keyedStateStore = keyedStateStore;
}
+ public void setAsyncKeyedStateStore(@Nullable AsyncKeyedStateStore asyncKeyedStateStore) {
Review Comment:
How about renaming `AsyncKeyedStateStore` to `KeyedStateStoreV2` or something? This is for new state v2 APIs instead of only for async states.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590604525
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
There are some considerations:
1. AsyncKeyedStateBackend should be initlized/setup with AEC, it's better to put the logic within AEC.
2. I prefer to let AEC control the life cycle of StateExecutor then operator could not know much about it.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1590503035
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -104,15 +105,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler {
public AsyncExecutionController(
MailboxExecutor mailboxExecutor,
- StateExecutor stateExecutor,
+ AsyncKeyedStateBackend asyncKeyedStateBackend,
int maxParallelism,
int batchSize,
long bufferTimeout,
int maxInFlightRecords) {
this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords);
this.mailboxExecutor = mailboxExecutor;
this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
- this.stateExecutor = stateExecutor;
+ if (asyncKeyedStateBackend != null) {
Review Comment:
Would it be better to put the initialization of the `state executor` in the `StreamOperator`?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java:
##########
@@ -242,6 +248,25 @@ private KeyedStateStore checkPreconditionsAndGetKeyedStateStore(
return keyedStateStore;
}
+ // TODO: Reconstruct this after StateManager is ready in FLIP-410.
Review Comment:
How about renaming it to `getValueState()`?
BTW, when the user provides the stateDescriptorV2, this method is used, and when the user provides the stateDescriptorV1, the old `getState` is used, right? Can stateDescriptorV2 and stateDescriptorV1 coexist?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackendBuilder.java:
##########
@@ -54,7 +59,14 @@ public class ForStKeyedStateBackendBuilder<K>
protected final Logger logger = LoggerFactory.getLogger(getClass());
+ private static final int KEY_SERIALIZER_BUFFER_START_SIZE = 32;
+
+ private static final int VALUE_SERIALIZER_BUFFER_START_SIZE = 128;
Review Comment:
Why is 128 chosen here?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35262] Bridge between AsyncKeyedStateBackend and AsyncExecutionController [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24740:
URL: https://github.com/apache/flink/pull/24740#discussion_r1592119008
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java:
##########
@@ -114,6 +116,10 @@ public void setKeyedStateStore(@Nullable KeyedStateStore keyedStateStore) {
this.keyedStateStore = keyedStateStore;
}
+ public void setAsyncKeyedStateStore(@Nullable AsyncKeyedStateStore asyncKeyedStateStore) {
Review Comment:
Thanks for the suggestion.
Renamed to `KeyedStateStoreV2`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org