You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2021/08/30 10:19:40 UTC
[flink] 02/02: [FLINK-23954][tests] Fix e2e test
test_queryable_state_restart_tm.sh
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4e50fe3d37753dc4dede7e047ba1b14fee71ca57
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Aug 27 20:23:47 2021 +0200
[FLINK-23954][tests] Fix e2e test test_queryable_state_restart_tm.sh
The problem was that we output the number of state entries at the time when the checkpoint
is confirmed. However, at this point in time we might have already added more elements to
the map. Hence, we risk reporting more elements than are actually contained in the checkpoint.
The solution is to remember what the element count was when creating the checkpoint.
This closes #17027.
---
.../tests/queryablestate/QsStateProducer.java | 47 ++++++++++++++--------
1 file changed, 30 insertions(+), 17 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 fee80e9..25281dc 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
@@ -27,20 +27,22 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.time.Duration;
import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Random;
/**
@@ -125,7 +127,8 @@ public class QsStateProducer {
new LabelSurrogate(LabelSurrogate.Type.values()[r % types], "bar");
synchronized (ctx.getCheckpointLock()) {
- ctx.collect(new Email(emailId, timestamp, foo, label));
+ final Email email = new Email(emailId, timestamp, foo, label);
+ ctx.collect(email);
}
Thread.sleep(30L);
@@ -139,17 +142,17 @@ public class QsStateProducer {
}
private static class TestFlatMap extends RichFlatMapFunction<Email, Object>
- implements CheckpointListener {
+ implements CheckpointListener, CheckpointedFunction {
private static final long serialVersionUID = 7821128115999005941L;
- private static final Logger LOG = LoggerFactory.getLogger(TestFlatMap.class);
-
private transient MapState<EmailId, EmailInformation> state;
private transient int count;
+ private transient Map<Long, Integer> countsAtCheckpoint;
+ private transient long lastCompletedCheckpoint;
@Override
- public void open(Configuration parameters) throws Exception {
+ public void open(Configuration parameters) {
MapStateDescriptor<EmailId, EmailInformation> stateDescriptor =
new MapStateDescriptor<>(
QsConstants.STATE_NAME,
@@ -158,24 +161,34 @@ public class QsStateProducer {
stateDescriptor.setQueryable(QsConstants.QUERY_NAME);
state = getRuntimeContext().getMapState(stateDescriptor);
- updateCount();
-
- LOG.info("Open {} with a count of {}.", getClass().getSimpleName(), count);
- }
-
- private void updateCount() throws Exception {
- count = Iterables.size(state.keys());
+ countsAtCheckpoint = new HashMap<>();
+ count = -1;
+ lastCompletedCheckpoint = -1;
}
@Override
public void flatMap(Email value, Collector<Object> out) throws Exception {
state.put(value.getEmailId(), new EmailInformation(value));
- updateCount();
+ count = Iterables.size(state.keys());
}
@Override
public void notifyCheckpointComplete(long checkpointId) {
- System.out.println("Count on snapshot: " + count); // we look for it in the test
+ if (checkpointId > lastCompletedCheckpoint) {
+ lastCompletedCheckpoint = checkpointId;
+ final int countAtCheckpoint = countsAtCheckpoint.remove(lastCompletedCheckpoint);
+
+ System.out.println(
+ "Count on snapshot: " + countAtCheckpoint); // we look for it in the test
+ }
}
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) {
+ countsAtCheckpoint.put(context.getCheckpointId(), count);
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) {}
}
}