[flink] branch master updated: [FLINK-23983][tests] Ensure to close and dispose created keyed-statebackend in StateBackendTestBase

[ta...@apache.org]

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

tangyun 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 4f5978f  [FLINK-23983][tests] Ensure to close and dispose created keyed-statebackend in StateBackendTestBase
4f5978f is described below

commit 4f5978fd4f53428b6af85f487f3f2abbf2f590fd
Author: Yun Tang <my...@live.com>
AuthorDate: Thu Sep 2 14:51:36 2021 +0800

    [FLINK-23983][tests] Ensure to close and dispose created keyed-statebackend in StateBackendTestBase
---
 .../flink/runtime/state/StateBackendTestBase.java  | 30 ++++++++++++++--------
 1 file changed, 19 insertions(+), 11 deletions(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
index 79e9da0..5867a4f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
@@ -150,6 +150,8 @@ import static org.mockito.Mockito.verify;
 /**
  * Tests for the {@link KeyedStateBackend} and {@link OperatorStateBackend} as produced by various
  * {@link StateBackend}s.
+ *
+ * <p>NOTE: Please ensure to close and dispose any created keyed state backend in tests.
  */
 @SuppressWarnings("serial")
 public abstract class StateBackendTestBase<B extends AbstractStateBackend> extends TestLogger {
@@ -3606,6 +3608,7 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
     public void testValueStateDefaultValue() throws Exception {
         ValueStateDescriptor<String> kvId = new ValueStateDescriptor<>("id", String.class, "Hello");
 
+        KeyedStateHandle keyedStateHandle;
         CheckpointableKeyedStateBackend<Integer> backend =
                 createKeyedBackend(IntSerializer.INSTANCE);
         try {
@@ -3622,17 +3625,22 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
             state.clear();
             assertEquals("Hello", state.value());
 
-            backend =
-                    restoreKeyedBackend(
-                            IntSerializer.INSTANCE,
-                            runSnapshot(
-                                    backend.snapshot(
-                                            1L,
-                                            1L,
-                                            createStreamFactory(),
-                                            CheckpointOptions.forCheckpointWithDefaultLocation()),
-                                    new SharedStateRegistry()));
-            state =
+            keyedStateHandle =
+                    runSnapshot(
+                            backend.snapshot(
+                                    1L,
+                                    1L,
+                                    createStreamFactory(),
+                                    CheckpointOptions.forCheckpointWithDefaultLocation()),
+                            new SharedStateRegistry());
+        } finally {
+            IOUtils.closeQuietly(backend);
+            backend.dispose();
+        }
+
+        try {
+            backend = restoreKeyedBackend(IntSerializer.INSTANCE, keyedStateHandle);
+            ValueState<String> state =
                     backend.getPartitionedState(
                             VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);