You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/06/15 03:53:20 UTC

[GitHub] [flink] Zakelly opened a new pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Zakelly opened a new pull request #16153:
URL: https://github.com/apache/flink/pull/16153


   ## What is the purpose of the change
   
   This change provides configurations as well as APIs to configure ChangelogStateBackend. Base on the discussion before (https://lists.apache.org/thread.html/ra178ce29088b1da362d98a5a6d8c7be48051caf1637ee24261738217%40%3Cdev.flink.apache.org%3E) and the design doc (https://docs.google.com/document/d/13AaCf5fczYTDHZ4G1mgYL685FqbnoEhgo0cdwuJlZmw/edit?usp=sharing, see Option 3), we make a refined version of loading ChangelogStateBackend. Please check the brief change log below.
   
   ## Brief change log
   
   - Define consistent override and combination policy (flag + state backend) in different config levels.
   - Differentiate explicitly the meaning of "enable flag" = true/false/unset.
   - Hide ChangelogStateBackend from users.
   
   
   ## Verifying this change
   
   This change added some test cases in ```ChangelogStateBackendLoadingTest```. Including tests for configuration overriding and stream graph validation.
   
   ## Does this pull request potentially affect one of the following parts:
   
     - Dependencies (does it add or upgrade a dependency): yes, tests in ```flink-statebackend-changelog``` now depends on ```flink-streaming-java_2.11```.
     - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes, add ```enableChangelogStateBackend``` and ```isChangelogStateBackendEnabled``` in ```StreamExecutionEnvironment```
     - The serializers: no
     - The runtime per-record code paths (performance sensitive): no, but ChangelogStateBackend itself may affect the per-record code paths.
     - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: yes
     - The S3 file system connector: no
    
   ## Documentation
     - Does this pull request introduce a new feature? yes.
     - If yes, how is the feature documented? Docs and JavaDocs.
   


-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118) 
   * 9ab9c0e4f4ba349c78d57596621e6771564cd88d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164) 
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654809893



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java
##########
@@ -175,21 +174,16 @@ public static CheckpointStorage load(
                     CheckpointingOptions.SAVEPOINT_DIRECTORY, defaultSavepointDirectory.toString());
         }
 
-        // Legacy state backends always take precedence for backwards compatibility.
-        StateBackend rootStateBackend =
-                (configuredStateBackend instanceof DelegatingStateBackend)
-                        ? ((DelegatingStateBackend) configuredStateBackend)
-                                .getDelegatedStateBackend()
-                        : configuredStateBackend;
-
-        if (rootStateBackend instanceof CheckpointStorage) {

Review comment:
       You are right. This should not be removed. I made a mistake.
   I revert this change. Thanks a lot for pointing this out.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r655169245



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,17 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
-                            "Whether to enable state backend to write state changes to StateChangelog.");
+                            "Whether to enable state backend to write state changes to StateChangelog. "
+                                    + "If this config is not set explicitly, it means no preference "
+                                    + "for enabling the change log, and the value in lower config "
+                                    + "level will take effect. The default value 'false' here means "
+                                    + "if no value set (job or cluster), the change log will not be"

Review comment:
       Yes, that's right. Thanks a lot for pointing this out! I have fixed this at [080b0ab](https://github.com/apache/flink/pull/16153/commits/080b0ab09afc18ee5e24bd8b536380273cae0d4c)
   
   Thanks again!




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118) 
   * 9ab9c0e4f4ba349c78d57596621e6771564cd88d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164) 
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118) 
   * 9ab9c0e4f4ba349c78d57596621e6771564cd88d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-863947779


   @Myasuka , thanks for the review! Would you mind taking another look? Thanks.


-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Myasuka commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r652468740



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
##########
@@ -327,6 +331,7 @@ private static StateBackend loadFromApplicationOrConfigOrDefaultInternal(
      */
     public static StateBackend fromApplicationOrConfigOrDefault(

Review comment:
       We should add logs to tell that whether changelog state backend is enabled.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -607,6 +625,65 @@ public StateBackend getStateBackend() {
         return defaultStateBackend;
     }
 
+    /**
+     * Enable the change log for current state backend. this changelog allows operators to persist
+     * state changes in a very fine-grained manner, as described below:
+     *
+     * <p>Stateful operators write the state changes to that log (logging the state), in addition to
+     * applying them to the state tables in RocksDB or the in-mem Hashtable.
+     *
+     * <p>An operator can acknowledge a checkpoint as soon as the changes in the log have reached
+     * the durable checkpoint storage.
+     *
+     * <p>The state tables are persisted periodically, independent of the checkpoints. We call this
+     * the materialization of the state on the checkpoint storage.
+     *
+     * <p>Once the state is materialized on checkpoint storage, the state changelog can be truncated
+     * to the corresponding point.
+     *
+     * <p>It establish a way to drastically reduce the checkpoint interval for streaming
+     * applications across state backends. For more details please check the FLIP-158.
+     *
+     * <p>If this method is not called explicitly, it means no preference for enabling the change
+     * log. Configs for change log enabling will override in different config levels
+     * (job/local/cluster).
+     *
+     * @param enabled true if enable the change log for state backend explicitly, otherwise disable
+     *     the change log.
+     * @return This StreamExecutionEnvironment itself, to allow chaining of function calls.
+     * @see #isChangelogStateBackendEnabled()
+     */
+    @PublicEvolving
+    public StreamExecutionEnvironment enableChangelogStateBackend(boolean enabled) {
+        if (enabled) {
+            try {
+                this.defaultStateBackend =
+                        StateBackendLoader.loadChangelogStateBackendIfNeeded(

Review comment:
       I don't think we should load state backend once we enable changelog state-backend as client-side might have different classloader with cluster-runtime-side. Just leave the original internal state backend and load it once running on cluster.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
##########
@@ -327,6 +331,7 @@ private static StateBackend loadFromApplicationOrConfigOrDefaultInternal(
      */
     public static StateBackend fromApplicationOrConfigOrDefault(
             @Nullable StateBackend fromApplication,
+            @Nullable TernaryBoolean isChangeLogStateBackendEnableFromApplication,

Review comment:
       If `isChangeLogStateBackendEnableFromApplication` is `null`, it actually has the meaning as `unset` (mainly on stream task side). However, `TernaryBoolean` itself is designed to be able to present the `unset` meaning. I think here existed mixed semantics and I suggest to remove the possibility of `Nullable`.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
##########
@@ -336,8 +341,18 @@ public static StateBackend fromApplicationOrConfigOrDefault(
                 loadFromApplicationOrConfigOrDefaultInternal(
                         fromApplication, config, classLoader, logger);
 
-        if (config.get(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG)
-                && !(fromApplication instanceof DelegatingStateBackend)) {
+        if (isChangeLogStateBackendEnableFromApplication == null) {
+            isChangeLogStateBackendEnableFromApplication = TernaryBoolean.UNDEFINED;
+        }
+        Optional<Boolean> enableConfig =
+                config.getOptional(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG);

Review comment:
       Why not use `Boolean enabledFromConfig = config.get(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG);` directly?

##########
File path: flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java
##########
@@ -183,16 +218,83 @@ public void testLoadingRocksDBStateBackend() throws Exception {
                 false);
     }
 
-    private Configuration config(String stateBackend) {
-        final Configuration config = config();
+    @Test
+    public void testConfigureEnvByConfiguration() throws Exception {

Review comment:
       If we decide to remove loading state backend in streaming execution environment, the methods of `testConfigureEnvByConfiguration` and `testEnableChangelogStateBackendInStreamExecutionEnvironment` are useless.

##########
File path: flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/BootstrapTransformation.java
##########
@@ -214,6 +215,8 @@ StreamConfig getConfig(
         config.setOperatorName(operatorID.toHexString());
         config.setOperatorID(operatorID);
         config.setStateBackend(stateBackend);
+        // This means leave this stateBackend unwrapped.

Review comment:
       ```suggestion
           // This means leaving this stateBackend unwrapped.
   ```

##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/utils/ExecutorUtils.java
##########
@@ -73,6 +75,7 @@ public static void setBatchProperties(StreamGraph streamGraph, TableConfig table
         // scheduler)
         streamGraph.setJobType(JobType.BATCH);
         streamGraph.setStateBackend(null);
+        streamGraph.setChangeLogStateBackendEnabled(TernaryBoolean.UNDEFINED);

Review comment:
       If no state-backend is needed for batch jobs, should we set set enable changlog as `TernaryBoolean.FALSE`?

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -593,7 +598,20 @@ public CheckpointingMode getCheckpointingMode() {
      */
     @PublicEvolving
     public StreamExecutionEnvironment setStateBackend(StateBackend backend) {
-        this.defaultStateBackend = Preconditions.checkNotNull(backend);
+        Preconditions.checkNotNull(backend);
+        if (changelogStateBackendEnabled.equals(TernaryBoolean.TRUE)) {
+            try {
+                this.defaultStateBackend =

Review comment:
       I don't think we should load the changelog state backend here, we could still keep the original internal state backend and load the real one within `StreamGraphGenerator` or `StreamTask`.
   Moreover, if so, there is no need to introduce `StateBackendLoader#loadChangelogStateBackendIfNeeded` and `StateBackendLoader#unwrapFromDelegatingStateBackend`.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Myasuka merged pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Myasuka merged pull request #16153:
URL: https://github.com/apache/flink/pull/16153


   


-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654809920



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
##########
@@ -384,6 +358,22 @@ public static boolean stateBackendFromApplicationOrConfigOrDefaultUseManagedMemo
         return false;
     }
 
+    /**
+     * Try to unwrap a DelegatingStateBackend and get the inner delegated StateBackend. If the
+     * provided StateBackend is not a delegated StateBackend, just return itself.
+     *
+     * @param backend the provided StateBackend that may delegate another StateBackend.
+     * @return the root StateBackend W/O delegation.
+     */
+    @VisibleForTesting
+    public static StateBackend unwrapFromDelegatingStateBackend(StateBackend backend) {
+        if (backend != null && backend instanceof DelegatingStateBackend) {

Review comment:
       Sure, moved to ```ChangelogStateBackendLoadingTest```.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 8a15ef20fa68b2af639336404483aafcb3f16515 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654810136



##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
##########
@@ -58,7 +58,14 @@
 
     private final StateBackend delegatedStateBackend;
 
-    public ChangelogStateBackend(StateBackend stateBackend) {
+    /**
+     * Delegate a state backend by a ChangelogStateBackend.
+     *
+     * <p>As FLINK-22678 mentioned, we currently hide this constructor from user.
+     *
+     * @param stateBackend the delegated state backend.
+     */
+    ChangelogStateBackend(StateBackend stateBackend) {

Review comment:
       Sure, I apply ```@Internal``` to class.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165) 
   * 1ef3786ca933d1bf675706d2f4f204314f3b540b UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-864496039


   @rkhachatryan thanks for the review! I've addressed your comments. Thanks!


-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     }, {
       "hash" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19215",
       "triggerID" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 1ef3786ca933d1bf675706d2f4f204314f3b540b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186) 
   * 080b0ab09afc18ee5e24bd8b536380273cae0d4c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19215) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068






-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861152936


   Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
   to review your pull request. We will use this comment to track the progress of the review.
   
   
   ## Automated Checks
   Last check on commit 080b0ab09afc18ee5e24bd8b536380273cae0d4c (Sat Aug 28 12:19:38 UTC 2021)
   
   **Warnings:**
    * **1 pom.xml files were touched**: Check for build and licensing issues.
    * No documentation files were touched! Remember to keep the Flink docs up to date!
   
   
   <sub>Mention the bot in a comment to re-run the automated checks.</sub>
   ## Review Progress
   
   * ❓ 1. The [description] looks good.
   * ❓ 2. There is [consensus] that the contribution should go into to Flink.
   * ❓ 3. Needs [attention] from.
   * ❓ 4. The change fits into the overall [architecture].
   * ❓ 5. Overall code [quality] is good.
   
   Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
    The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
    - `@flinkbot approve all` to approve all aspects
    - `@flinkbot approve-until architecture` to approve everything until `architecture`
    - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
    - `@flinkbot disapprove architecture` to remove an approval you gave earlier
   </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

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 8a15ef20fa68b2af639336404483aafcb3f16515 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] rkhachatryan commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654952839



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -607,6 +612,53 @@ public StateBackend getStateBackend() {
         return defaultStateBackend;
     }
 
+    /**
+     * Enable the change log for current state backend. this changelog allows operators to persist
+     * state changes in a very fine-grained manner, as described below:
+     *
+     * <p>Stateful operators write the state changes to that log (logging the state), in addition to
+     * applying them to the state tables in RocksDB or the in-mem Hashtable.
+     *
+     * <p>An operator can acknowledge a checkpoint as soon as the changes in the log have reached
+     * the durable checkpoint storage.
+     *
+     * <p>The state tables are persisted periodically, independent of the checkpoints. We call this
+     * the materialization of the state on the checkpoint storage.
+     *
+     * <p>Once the state is materialized on checkpoint storage, the state changelog can be truncated
+     * to the corresponding point.
+     *
+     * <p>It establish a way to drastically reduce the checkpoint interval for streaming
+     * applications across state backends. For more details please check the FLIP-158.
+     *
+     * <p>If this method is not called explicitly, it means no preference for enabling the change
+     * log. Configs for change log enabling will override in different config levels
+     * (job/local/cluster).
+     *
+     * @param enabled true if enable the change log for state backend explicitly, otherwise disable
+     *     the change log.
+     * @return This StreamExecutionEnvironment itself, to allow chaining of function calls.
+     * @see #isChangelogStateBackendEnabled()
+     */
+    @PublicEvolving
+    public StreamExecutionEnvironment enableChangelogStateBackend(boolean enabled) {

Review comment:
       Thanks! Could you please do the same for the scala version? 
   Sorry for clarifying in the beginning.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "CANCELED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 9ab9c0e4f4ba349c78d57596621e6771564cd88d Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164) 
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r655169245



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,17 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
-                            "Whether to enable state backend to write state changes to StateChangelog.");
+                            "Whether to enable state backend to write state changes to StateChangelog. "
+                                    + "If this config is not set explicitly, it means no preference "
+                                    + "for enabling the change log, and the value in lower config "
+                                    + "level will take effect. The default value 'false' here means "
+                                    + "if no value set (job or cluster), the change log will not be"

Review comment:
       Yes, that's right. Thanks a lot for pointing this out! I have fixed this at [080b0ab](https://github.com/apache/flink/pull/16153/commits/080b0ab09afc18ee5e24bd8b536380273cae0d4c)
   
   Thanks again!




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Myasuka commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654247587



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,15 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
-                    .defaultValue(false)
+                    .noDefaultValue()

Review comment:
       As discussed offline, this configuration should have a default value of `false`. We might have different kinds of implementation but we should tell user explicitly that this option is disabled by default.




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654332187



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,15 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
-                    .defaultValue(false)
+                    .noDefaultValue()

Review comment:
       Yes, the default value here is false, but the return value of ```StreamExecutionEnvironment::isChangelogStateBackendEnabled``` is a ```TernaryBoolean```, which may bring some confusion. The ```TernaryBoolean``` is essential cause table planner uses the value to create StreamGraphGenerator (see [here](https://github.com/apache/flink/pull/16153/commits/2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177#diff-c2ed4f07637221036620a06234633cbed96eedf4f29297fd3d766b3e1e751cbaR48)). @rkhachatryan and @curcur would you please to take a look?

##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,15 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
-                    .defaultValue(false)
+                    .noDefaultValue()

Review comment:
       Yes, I set the default value here to false, but the return value of ```StreamExecutionEnvironment::isChangelogStateBackendEnabled``` is a ```TernaryBoolean```, which may bring some confusion. The ```TernaryBoolean``` is essential cause table planner uses the value to create StreamGraphGenerator (see [here](https://github.com/apache/flink/pull/16153/commits/2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177#diff-c2ed4f07637221036620a06234633cbed96eedf4f29297fd3d766b3e1e751cbaR48)). @rkhachatryan and @curcur would you please to take a look?




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot commented on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861152936


   Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
   to review your pull request. We will use this comment to track the progress of the review.
   
   
   ## Automated Checks
   Last check on commit 8a15ef20fa68b2af639336404483aafcb3f16515 (Tue Jun 15 04:07:51 UTC 2021)
   
   **Warnings:**
    * **1 pom.xml files were touched**: Check for build and licensing issues.
    * No documentation files were touched! Remember to keep the Flink docs up to date!
   
   
   <sub>Mention the bot in a comment to re-run the automated checks.</sub>
   ## Review Progress
   
   * ❓ 1. The [description] looks good.
   * ❓ 2. There is [consensus] that the contribution should go into to Flink.
   * ❓ 3. Needs [attention] from.
   * ❓ 4. The change fits into the overall [architecture].
   * ❓ 5. Overall code [quality] is good.
   
   Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
    The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
    - `@flinkbot approve all` to approve all aspects
    - `@flinkbot approve-until architecture` to approve everything until `architecture`
    - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
    - `@flinkbot disapprove architecture` to remove an approval you gave earlier
   </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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhangyy91 commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
zhangyy91 commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r655154815



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,17 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
-                            "Whether to enable state backend to write state changes to StateChangelog.");
+                            "Whether to enable state backend to write state changes to StateChangelog. "
+                                    + "If this config is not set explicitly, it means no preference "
+                                    + "for enabling the change log, and the value in lower config "
+                                    + "level will take effect. The default value 'false' here means "
+                                    + "if no value set (job or cluster), the change log will not be"

Review comment:
       @Zakelly Hi, I reviewed the pull request and, in general LGTM.
   I found a minor error in the doc string. There should be a space at the end of the line. Or the generated doc will be "the change log will not beenabled".
   
   




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "FAILURE",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118) 
   * 9ab9c0e4f4ba349c78d57596621e6771564cd88d UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068






-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654810546



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -607,6 +612,53 @@ public StateBackend getStateBackend() {
         return defaultStateBackend;
     }
 
+    /**
+     * Enable the change log for current state backend. this changelog allows operators to persist
+     * state changes in a very fine-grained manner, as described below:
+     *
+     * <p>Stateful operators write the state changes to that log (logging the state), in addition to
+     * applying them to the state tables in RocksDB or the in-mem Hashtable.
+     *
+     * <p>An operator can acknowledge a checkpoint as soon as the changes in the log have reached
+     * the durable checkpoint storage.
+     *
+     * <p>The state tables are persisted periodically, independent of the checkpoints. We call this
+     * the materialization of the state on the checkpoint storage.
+     *
+     * <p>Once the state is materialized on checkpoint storage, the state changelog can be truncated
+     * to the corresponding point.
+     *
+     * <p>It establish a way to drastically reduce the checkpoint interval for streaming
+     * applications across state backends. For more details please check the FLIP-158.
+     *
+     * <p>If this method is not called explicitly, it means no preference for enabling the change
+     * log. Configs for change log enabling will override in different config levels
+     * (job/local/cluster).
+     *
+     * @param enabled true if enable the change log for state backend explicitly, otherwise disable
+     *     the change log.
+     * @return This StreamExecutionEnvironment itself, to allow chaining of function calls.
+     * @see #isChangelogStateBackendEnabled()
+     */
+    @PublicEvolving
+    public StreamExecutionEnvironment enableChangelogStateBackend(boolean enabled) {

Review comment:
       Yes, I added some description. 🙂




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     }, {
       "hash" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 1ef3786ca933d1bf675706d2f4f204314f3b540b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186) 
   * 080b0ab09afc18ee5e24bd8b536380273cae0d4c UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Zakelly commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Zakelly commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r655046921



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -607,6 +612,53 @@ public StateBackend getStateBackend() {
         return defaultStateBackend;
     }
 
+    /**
+     * Enable the change log for current state backend. this changelog allows operators to persist
+     * state changes in a very fine-grained manner, as described below:
+     *
+     * <p>Stateful operators write the state changes to that log (logging the state), in addition to
+     * applying them to the state tables in RocksDB or the in-mem Hashtable.
+     *
+     * <p>An operator can acknowledge a checkpoint as soon as the changes in the log have reached
+     * the durable checkpoint storage.
+     *
+     * <p>The state tables are persisted periodically, independent of the checkpoints. We call this
+     * the materialization of the state on the checkpoint storage.
+     *
+     * <p>Once the state is materialized on checkpoint storage, the state changelog can be truncated
+     * to the corresponding point.
+     *
+     * <p>It establish a way to drastically reduce the checkpoint interval for streaming
+     * applications across state backends. For more details please check the FLIP-158.
+     *
+     * <p>If this method is not called explicitly, it means no preference for enabling the change
+     * log. Configs for change log enabling will override in different config levels
+     * (job/local/cluster).
+     *
+     * @param enabled true if enable the change log for state backend explicitly, otherwise disable
+     *     the change log.
+     * @return This StreamExecutionEnvironment itself, to allow chaining of function calls.
+     * @see #isChangelogStateBackendEnabled()
+     */
+    @PublicEvolving
+    public StreamExecutionEnvironment enableChangelogStateBackend(boolean enabled) {

Review comment:
       Sure thing, at [1ef3786](https://github.com/apache/flink/pull/16153/commits/1ef3786ca933d1bf675706d2f4f204314f3b540b).
   And as for the python datastream API, I left it unimplemented. I'll create another PR for 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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] rkhachatryan commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r654583087



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
##########
@@ -384,6 +358,22 @@ public static boolean stateBackendFromApplicationOrConfigOrDefaultUseManagedMemo
         return false;
     }
 
+    /**
+     * Try to unwrap a DelegatingStateBackend and get the inner delegated StateBackend. If the
+     * provided StateBackend is not a delegated StateBackend, just return itself.
+     *
+     * @param backend the provided StateBackend that may delegate another StateBackend.
+     * @return the root StateBackend W/O delegation.
+     */
+    @VisibleForTesting
+    public static StateBackend unwrapFromDelegatingStateBackend(StateBackend backend) {
+        if (backend != null && backend instanceof DelegatingStateBackend) {

Review comment:
       nit: Is this method used **only** for tests? If so, maybe move it to `test` source directory?
   nit: I think `null` check is unnecessary here.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
##########
@@ -607,6 +612,53 @@ public StateBackend getStateBackend() {
         return defaultStateBackend;
     }
 
+    /**
+     * Enable the change log for current state backend. this changelog allows operators to persist
+     * state changes in a very fine-grained manner, as described below:
+     *
+     * <p>Stateful operators write the state changes to that log (logging the state), in addition to
+     * applying them to the state tables in RocksDB or the in-mem Hashtable.
+     *
+     * <p>An operator can acknowledge a checkpoint as soon as the changes in the log have reached
+     * the durable checkpoint storage.
+     *
+     * <p>The state tables are persisted periodically, independent of the checkpoints. We call this
+     * the materialization of the state on the checkpoint storage.
+     *
+     * <p>Once the state is materialized on checkpoint storage, the state changelog can be truncated
+     * to the corresponding point.
+     *
+     * <p>It establish a way to drastically reduce the checkpoint interval for streaming
+     * applications across state backends. For more details please check the FLIP-158.
+     *
+     * <p>If this method is not called explicitly, it means no preference for enabling the change
+     * log. Configs for change log enabling will override in different config levels
+     * (job/local/cluster).
+     *
+     * @param enabled true if enable the change log for state backend explicitly, otherwise disable
+     *     the change log.
+     * @return This StreamExecutionEnvironment itself, to allow chaining of function calls.
+     * @see #isChangelogStateBackendEnabled()
+     */
+    @PublicEvolving
+    public StreamExecutionEnvironment enableChangelogStateBackend(boolean enabled) {

Review comment:
       very good description :+1: :slightly_smiling_face: 
   
   Should we also mention that it only applies to keyed state currently? (so non-keyed operator state and channel state are checkpointed as usual)

##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
##########
@@ -58,7 +58,14 @@
 
     private final StateBackend delegatedStateBackend;
 
-    public ChangelogStateBackend(StateBackend stateBackend) {
+    /**
+     * Delegate a state backend by a ChangelogStateBackend.
+     *
+     * <p>As FLINK-22678 mentioned, we currently hide this constructor from user.
+     *
+     * @param stateBackend the delegated state backend.
+     */
+    ChangelogStateBackend(StateBackend stateBackend) {

Review comment:
       Maybe also mark the class as `@Internal` to make the intention clear to the user?

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java
##########
@@ -175,21 +174,16 @@ public static CheckpointStorage load(
                     CheckpointingOptions.SAVEPOINT_DIRECTORY, defaultSavepointDirectory.toString());
         }
 
-        // Legacy state backends always take precedence for backwards compatibility.
-        StateBackend rootStateBackend =
-                (configuredStateBackend instanceof DelegatingStateBackend)
-                        ? ((DelegatingStateBackend) configuredStateBackend)
-                                .getDelegatedStateBackend()
-                        : configuredStateBackend;
-
-        if (rootStateBackend instanceof CheckpointStorage) {

Review comment:
       Could you please explain the removal? 
   IIUC, changelog state backend can still be passed here (even if hidden from user) when called from `StreamTask#createCheckpointStorage`
   
   cc: @curcur 




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] Myasuka merged pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
Myasuka merged pull request #16153:
URL: https://github.com/apache/flink/pull/16153


   


-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot commented on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 8a15ef20fa68b2af639336404483aafcb3f16515 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 50b2100f9a62e662d6cb4a7ec111d3e73b9cd843 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165) 
   * 1ef3786ca933d1bf675706d2f4f204314f3b540b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] zhangyy91 commented on a change in pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
zhangyy91 commented on a change in pull request #16153:
URL: https://github.com/apache/flink/pull/16153#discussion_r655154815



##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -102,13 +102,17 @@
 
     /** Whether to enable state change log. */
     @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS)
-    @Documentation.ExcludeFromDocumentation("Hidden for now")
     public static final ConfigOption<Boolean> ENABLE_STATE_CHANGE_LOG =
             ConfigOptions.key("state.backend.changelog.enabled")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
-                            "Whether to enable state backend to write state changes to StateChangelog.");
+                            "Whether to enable state backend to write state changes to StateChangelog. "
+                                    + "If this config is not set explicitly, it means no preference "
+                                    + "for enabling the change log, and the value in lower config "
+                                    + "level will take effect. The default value 'false' here means "
+                                    + "if no value set (job or cluster), the change log will not be"

Review comment:
       @Zakelly Hi, I reviewed the pull request and, in general LGTM.
   I found a minor error in the doc string. There should be a space at the end of the line. Or the generated doc will be "the change log will not beenabled".
   
   




-- 
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 1ef3786ca933d1bf675706d2f4f204314f3b540b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [flink] flinkbot edited a comment on pull request #16153: [FLINK-22678] Configurations and user APIs for ChangelogStateBackend

Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16153:
URL: https://github.com/apache/flink/pull/16153#issuecomment-861162068


   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18971",
       "triggerID" : "8a15ef20fa68b2af639336404483aafcb3f16515",
       "triggerType" : "PUSH"
     }, {
       "hash" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19071",
       "triggerID" : "91f21adf6e0239eec7f093e767c85cdf9acd1f3d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19118",
       "triggerID" : "2c78381d58d1e7e3ee3edd06afb03e7e6eb1f177",
       "triggerType" : "PUSH"
     }, {
       "hash" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19164",
       "triggerID" : "9ab9c0e4f4ba349c78d57596621e6771564cd88d",
       "triggerType" : "PUSH"
     }, {
       "hash" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19165",
       "triggerID" : "50b2100f9a62e662d6cb4a7ec111d3e73b9cd843",
       "triggerType" : "PUSH"
     }, {
       "hash" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19186",
       "triggerID" : "1ef3786ca933d1bf675706d2f4f204314f3b540b",
       "triggerType" : "PUSH"
     }, {
       "hash" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19215",
       "triggerID" : "080b0ab09afc18ee5e24bd8b536380273cae0d4c",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 080b0ab09afc18ee5e24bd8b536380273cae0d4c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19215) 
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run travis` re-run the last Travis build
    - `@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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org