[flink] branch master updated: [hotfix][state backend] Put the delegating log to ChangelogStateBackend constructor

[roman]

This is an automated email from the ASF dual-hosted git repository.

roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new de9cf86  [hotfix][state backend] Put the delegating log to 
ChangelogStateBackend constructor
de9cf86 is described below

commit de9cf867665644224c22ea5d2624afadcf3caa1e
Author: Yuan Mei <yuanmei.w...@gmail.com>
AuthorDate: Fri Mar 19 15:03:35 2021 +0800

    [hotfix][state backend] Put the delegating log to ChangelogStateBackend 
constructor
---
 .../java/org/apache/flink/runtime/state/StateBackendLoader.java  | 4 ----
 .../org/apache/flink/state/changelog/ChangelogStateBackend.java  | 9 +++++++++
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
index 75a2b62..4e1ca3a 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java
@@ -386,10 +386,6 @@ public class StateBackendLoader {
     private static StateBackend loadChangelogStateBackend(
             StateBackend backend, ClassLoader classLoader) throws 
DynamicCodeLoadingException {
 
-        LOG.info(
-                "Delegate State Backend is used, and the root State Backend is 
{}",
-                backend.getClass().getSimpleName());
-
         // ChangelogStateBackend resides in a separate module, load it using 
reflection
         try {
             Constructor<? extends DelegatingStateBackend> constructor =
diff --git 
a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
 
b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
index a855ad8..f1192d9 100644
--- 
a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
+++ 
b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogStateBackend.java
@@ -38,6 +38,9 @@ import 
org.apache.flink.runtime.state.delegate.DelegatingStateBackend;
 import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
 import org.apache.flink.util.Preconditions;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nonnull;
 
 import java.util.Collection;
@@ -50,6 +53,8 @@ public class ChangelogStateBackend implements 
DelegatingStateBackend, Configurab
 
     private static final long serialVersionUID = 1000L;
 
+    private static final Logger LOG = 
LoggerFactory.getLogger(ChangelogStateBackend.class);
+
     private final StateBackend delegatedStateBackend;
 
     public ChangelogStateBackend(StateBackend stateBackend) {
@@ -58,6 +63,10 @@ public class ChangelogStateBackend implements 
DelegatingStateBackend, Configurab
         Preconditions.checkArgument(
                 !(stateBackend instanceof DelegatingStateBackend),
                 "Recursive Delegation is not supported.");
+
+        LOG.info(
+                "ChangelogStateBackend is used, delegating {}.",
+                delegatedStateBackend.getClass().getSimpleName());
     }
 
     @Override