[flink] branch release-1.13 updated: [FLINK-23097][tests] Harden test_queryable_state_restart_tm.sh

[tr...@apache.org]

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

trohrmann pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.13 by this push:
     new 7522bd7  [FLINK-23097][tests] Harden test_queryable_state_restart_tm.sh
7522bd7 is described below

commit 7522bd71cca80dcfd69f64e20020c174073d3ff3
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Aug 20 12:50:16 2021 +0200

    [FLINK-23097][tests] Harden test_queryable_state_restart_tm.sh
    
    This commit hardens the test_queryable_state_restart_tm.sh e2e test by only writing the
    count value of state entries on notifyCheckpointComplete. Before we have written out uncommitted
    information out that led to test failures.
    
    This closes #16910.
---
 .../streaming/tests/queryablestate/QsStateProducer.java      | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/QsStateProducer.java b/flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/QsStateProducer.java
index 3200185..5cbaa97 100644
--- a/flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/QsStateProducer.java
+++ b/flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/QsStateProducer.java
@@ -18,6 +18,7 @@
 package org.apache.flink.streaming.tests.queryablestate;
 
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.state.CheckpointListener;
 import org.apache.flink.api.common.state.MapState;
 import org.apache.flink.api.common.state.MapStateDescriptor;
 import org.apache.flink.api.common.typeinfo.TypeHint;
@@ -138,7 +139,7 @@ public class QsStateProducer {
     }
 
     private static class TestFlatMap extends RichFlatMapFunction<Email, Object>
-            implements CheckpointedFunction {
+            implements CheckpointedFunction, CheckpointListener {
 
         private static final long serialVersionUID = 7821128115999005941L;
 
@@ -165,11 +166,14 @@ public class QsStateProducer {
         }
 
         @Override
-        public void snapshotState(FunctionSnapshotContext context) {
-            System.out.println("Count on snapshot: " + count); // we look for it in the test
-        }
+        public void snapshotState(FunctionSnapshotContext context) {}
 
         @Override
         public void initializeState(FunctionInitializationContext context) {}
+
+        @Override
+        public void notifyCheckpointComplete(long checkpointId) throws Exception {
+            System.out.println("Count on snapshot: " + count); // we look for it in the test
+        }
     }
 }