You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Zakelly (via GitHub)" <gi...@apache.org> on 2024/10/23 07:02:25 UTC
[PR] [FLINK-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Zakelly opened a new pull request, #25560:
URL: https://github.com/apache/flink/pull/25560
## What is the purpose of the change
Currently, the operator will create both `AsyncKeyedStateBackend` and `KeyedStateBackend` and pick one for use. This PR only creates one depending on whether the operator performs async state processing or not.
## Brief change log
- Introduce internal method `isAsyncStateProcessingEnabled` in `AbstractStreamOperator`.
- Determine `isAsyncStateProcessingEnabled` in state initialization (including timerservice).
- Remove redundant override methods.
## Verifying this change
This change is already covered by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25560:
URL: https://github.com/apache/flink/pull/25560#issuecomment-2434263848
CI failure addressed by #25562 , rebasing...
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25560:
URL: https://github.com/apache/flink/pull/25560#issuecomment-2431092876
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a68f2b281760b51ad5ab73f2938f95dc7c29e0be",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a68f2b281760b51ad5ab73f2938f95dc7c29e0be",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a68f2b281760b51ad5ab73f2938f95dc7c29e0be 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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #25560:
URL: https://github.com/apache/flink/pull/25560#discussion_r1812428234
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java:
##########
@@ -252,31 +263,32 @@ public StreamOperatorStateContext streamOperatorStateContext(
(prioritizedOperatorSubtaskStates.isRestored() && !isUsingCustomRawKeyedState)
? rawKeyedStateInputs
: Collections.emptyList();
- if (keyedStatedBackend != null) {
- timeServiceManager =
- timeServiceManagerProvider.create(
- environment.getMetricGroup().getIOMetricGroup(),
- keyedStatedBackend,
- keyedStatedBackend.getKeyGroupRange(),
- environment.getUserCodeClassLoader().asClassLoader(),
- keyContext,
- processingTimeService,
- restoredRawKeyedStateTimers,
- cancellationContext);
- }
- if (stateBackend.supportsAsyncKeyedStateBackend()) {
- asyncTimeServiceManager =
- timeServiceManagerProvider.create(
- environment.getMetricGroup().getIOMetricGroup(),
- asyncKeyedStateBackend,
- asyncKeyedStateBackend.getKeyGroupRange(),
- environment.getUserCodeClassLoader().asClassLoader(),
- keyContext,
- processingTimeService,
- restoredRawKeyedStateTimers,
- cancellationContext);
+ if (isAsyncState) {
+ if (asyncKeyedStateBackend != null) {
Review Comment:
If `asyncKeyedStateBackend` is instance of `adaptor`, is it better to use the delegated `keyedStateBackend` to create `timerManager`?
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly merged PR #25560:
URL: https://github.com/apache/flink/pull/25560
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25560:
URL: https://github.com/apache/flink/pull/25560#issuecomment-2437562920
@flinkbot run azure
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25560:
URL: https://github.com/apache/flink/pull/25560#discussion_r1812449913
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java:
##########
@@ -252,31 +263,32 @@ public StreamOperatorStateContext streamOperatorStateContext(
(prioritizedOperatorSubtaskStates.isRestored() && !isUsingCustomRawKeyedState)
? rawKeyedStateInputs
: Collections.emptyList();
- if (keyedStatedBackend != null) {
- timeServiceManager =
- timeServiceManagerProvider.create(
- environment.getMetricGroup().getIOMetricGroup(),
- keyedStatedBackend,
- keyedStatedBackend.getKeyGroupRange(),
- environment.getUserCodeClassLoader().asClassLoader(),
- keyContext,
- processingTimeService,
- restoredRawKeyedStateTimers,
- cancellationContext);
- }
- if (stateBackend.supportsAsyncKeyedStateBackend()) {
- asyncTimeServiceManager =
- timeServiceManagerProvider.create(
- environment.getMetricGroup().getIOMetricGroup(),
- asyncKeyedStateBackend,
- asyncKeyedStateBackend.getKeyGroupRange(),
- environment.getUserCodeClassLoader().asClassLoader(),
- keyContext,
- processingTimeService,
- restoredRawKeyedStateTimers,
- cancellationContext);
+ if (isAsyncState) {
+ if (asyncKeyedStateBackend != null) {
Review Comment:
In this case, the `AsyncKeyedStateBackendAdaptor#create` is invoked. So yes, this is the current behavior.
--
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-36589][state/runtime] Decouple the initialization of sync and async keyed statebackend [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25560:
URL: https://github.com/apache/flink/pull/25560#issuecomment-2431172226
@flinkbot run azure
--
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