[FLINK-38462] Introduce state backend type identifier

Author: rkhachatryan

flink-core/src/main/java/org/apache/flink/api/common/state/KeyedStateStore.java

@@ -338,4 +338,11 @@ org.apache.flink.api.common.state.v2.AggregatingState<IN, OUT> getAggregatingSta
                             org.apache.flink.api.common.state.v2.AggregatingStateDescriptor<
                                             IN, ACC, OUT>
                                     stateProperties);
+
+    /**
+     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
+     *     rocksdb, hashmap, forst, batch.
+     */
+    @Experimental
+    String getBackendTypeIdentifier();
 }

flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/utils/TestSharedBuffer.java

@@ -287,6 +287,11 @@ org.apache.flink.api.common.state.v2.AggregatingState<IN, OUT> getAggregatingSta
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public String getBackendTypeIdentifier() {
+            return "mock";
+        }
+
         private class CountingIterator<T> implements Iterator<T> {
 
             private final Iterator<T> iterator;

flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java

@@ -324,6 +324,11 @@ public <N> Stream<Tuple2<Integer, N>> getKeysAndNamespaces(String state) {
                     "Operations other than getKeys() are not supported on this testing StateBackend.");
         }
 
+        @Override
+        public String getBackendTypeIdentifier() {
+            return "test";
+        }
+
         @Nonnull
         @Override
         public SavepointResources<Integer> savepoint() throws UnsupportedOperationException {

flink-runtime/src/main/java/org/apache/flink/runtime/state/AsyncKeyedStateBackend.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.state;
 
+import org.apache.flink.annotation.Experimental;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.state.InternalCheckpointListener;
 import org.apache.flink.api.common.state.v2.State;
@@ -140,4 +141,11 @@ default boolean isSafeToReuseKVState() {
 
     @Override
     void dispose();
+
+    /**
+     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
+     *     rocksdb, hashmap, or unknown.
+     */
+    @Experimental
+    String getBackendTypeIdentifier();
 }

flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java

@@ -219,6 +219,15 @@ org.apache.flink.api.common.state.v2.AggregatingState<IN, OUT> getAggregatingSta
         }
     }
 
+    @Override
+    public String getBackendTypeIdentifier() {
+        if (keyedStateBackend != null) {
+            return keyedStateBackend.getBackendTypeIdentifier();
+        } else {
+            return asyncKeyedStateBackend.getBackendTypeIdentifier();
+        }
+    }
+
     protected <S extends org.apache.flink.api.common.state.v2.State, SV> S getPartitionedState(
             org.apache.flink.api.common.state.v2.StateDescriptor<SV> stateDescriptor)
             throws Exception {

flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.state;
 
+import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.state.State;
 import org.apache.flink.api.common.state.StateDescriptor;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -166,6 +167,13 @@ default boolean isSafeToReuseKVState() {
         return false;
     }
 
+    /**
+     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
+     *     rocksdb, hashmap, forst, batch.
+     */
+    @Experimental
+    String getBackendTypeIdentifier();
+
     /** Listener is given a callback when {@link #setCurrentKey} is called (key context changes). */
     @FunctionalInterface
     interface KeySelectionListener<K> {

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

@@ -46,6 +46,7 @@
 import org.apache.flink.runtime.state.SnapshotResult;
 import org.apache.flink.runtime.state.SnapshotStrategy;
 import org.apache.flink.runtime.state.SnapshotStrategyRunner;
+import org.apache.flink.runtime.state.StateBackendLoader;
 import org.apache.flink.runtime.state.StateEntry;
 import org.apache.flink.runtime.state.StateSnapshotRestore;
 import org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory;
@@ -406,6 +407,11 @@ public <N> Stream<Tuple2<K, N>> getKeysAndNamespaces(String state) {
         return table.getKeysAndNamespaces();
     }
 
+    @Override
+    public String getBackendTypeIdentifier() {
+        return StateBackendLoader.HASHMAP_STATE_BACKEND_NAME;
+    }
+
     @Override
     @Nonnull
     public <N, SV, SEV, S extends State, IS extends S> IS createOrUpdateInternalState(

flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/AsyncKeyedStateBackendAdaptor.java

@@ -132,6 +132,11 @@ public void switchContext(@Nullable RecordContext<K> context) {
     @Override
     public void dispose() {}
 
+    @Override
+    public String getBackendTypeIdentifier() {
+        return keyedStateBackend.getBackendTypeIdentifier();
+    }
+
     @Override
     public void close() throws IOException {}
 

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java

@@ -229,6 +229,11 @@ public boolean deregisterKeySelectionListener(KeySelectionListener<K> listener)
         return keySelectionListeners.remove(listener);
     }
 
+    @Override
+    public String getBackendTypeIdentifier() {
+        return "batch";
+    }
+
     @Nonnull
     @Override
     public <N, SV, SEV, S extends State, IS extends S> IS createOrUpdateInternalState(

flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestUtils.java

@@ -171,6 +171,11 @@ public void dispose() {
             // do nothing
         }
 
+        @Override
+        public String getBackendTypeIdentifier() {
+            return "test";
+        }
+
         @Override
         public void close() {
             // do nothing
@@ -328,6 +333,11 @@ public void dispose() {
                     delegatedKeyedStateBackend.dispose();
                 }
 
+                @Override
+                public String getBackendTypeIdentifier() {
+                    return "test";
+                }
+
                 @Override
                 public void close() throws IOException {
                     super.close();