You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2020/10/22 08:18:14 UTC
[flink] branch release-1.11 updated (ffe790f -> 4bff79f)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from ffe790f [hotfix][cep][docs] Fix the documentation of cep sql
new 4da26b8 [tests][checkpointing] Test retrieval error fails ZKStore.recover
new 4bff79f [FLINK-19401][checkpointing] Download checkpoints only if needed
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../ZooKeeperCompletedCheckpointStore.java | 24 +-
.../checkpoint/CompletedCheckpointStoreTest.java | 5 +-
.../ZooKeeperCompletedCheckpointStoreTest.java | 111 +++++
.../ZooKeeperHighAvailabilityITCase.java | 456 ---------------------
4 files changed, 135 insertions(+), 461 deletions(-)
delete mode 100644 flink-tests/src/test/java/org/apache/flink/test/checkpointing/ZooKeeperHighAvailabilityITCase.java
[flink] 02/02: [FLINK-19401][checkpointing] Download checkpoints
only if needed
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4bff79f5efffccd1793e09a5e08d0ceb9fe90cf2
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Oct 20 17:02:41 2020 +0200
[FLINK-19401][checkpointing] Download checkpoints only if needed
If checkpoints in ZK are the same as already downloaded then skip
download. This prevents excessive S3 reads and eventually failure to
start the job, particularly in multi-region recovery.
ZooKeeperHighAvailabilityITCase is dropped is it now fails
false-positively. Instead, unit testRetrievalFailsRecovery
was added in the previous commit.
---
.../ZooKeeperCompletedCheckpointStore.java | 19 +
.../ZooKeeperCompletedCheckpointStoreTest.java | 33 ++
.../ZooKeeperHighAvailabilityITCase.java | 456 ---------------------
3 files changed, 52 insertions(+), 456 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
index f52d5eb..1b545f9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
@@ -36,7 +36,9 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.ConcurrentModificationException;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -144,6 +146,10 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
int numberOfInitialCheckpoints = initialCheckpoints.size();
LOG.info("Found {} checkpoints in ZooKeeper.", numberOfInitialCheckpoints);
+ if (haveAllDownloaded(initialCheckpoints)) {
+ LOG.info("All {} checkpoints found are already downloaded.", numberOfInitialCheckpoints);
+ return;
+ }
// Try and read the state handles from storage. We try until we either successfully read
// all of them or when we reach a stable state, i.e. when we successfully read the same set
@@ -204,6 +210,19 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
}
}
+ private boolean haveAllDownloaded(List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> checkpointPointers) {
+ if (completedCheckpoints.size() != checkpointPointers.size()) {
+ return false;
+ }
+ Set<Long> localIds = completedCheckpoints.stream().map(CompletedCheckpoint::getCheckpointID).collect(Collectors.toSet());
+ for (Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String> initialCheckpoint : checkpointPointers) {
+ if (!localIds.contains(pathToCheckpointId(initialCheckpoint.f1))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Synchronously writes the new checkpoints to ZooKeeper and asynchronously removes older ones.
*
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
index 503dc90..0042293 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
@@ -44,7 +44,9 @@ import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
+import java.util.stream.IntStream;
+import static org.apache.flink.runtime.checkpoint.CompletedCheckpointStoreTest.createCheckpoint;
import static org.apache.flink.util.ExceptionUtils.findThrowable;
import static org.apache.flink.util.ExceptionUtils.rethrow;
import static org.junit.Assert.assertEquals;
@@ -82,6 +84,37 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger {
});
}
+ @Test
+ public void testNoDownloadIfCheckpointsNotChanged() throws Exception {
+ testDownloadInternal((store, checkpointsInZk, sharedStateRegistry) -> {
+ try {
+ checkpointsInZk.add(createHandle(1, id -> {
+ throw new AssertionError("retrieveState was attempted for checkpoint " + id);
+ }));
+ store.addCheckpoint(createCheckpoint(1, sharedStateRegistry));
+ store.recover(); // will fail in case of attempt to retrieve state
+ } catch (Exception exception) {
+ throw new RuntimeException(exception);
+ }
+ });
+ }
+
+ @Test
+ public void testDownloadIfCheckpointsChanged() throws Exception {
+ testDownloadInternal((store, checkpointsInZk, sharedStateRegistry) -> {
+ try {
+ int lastInZk = 10;
+ IntStream.range(0, lastInZk + 1).forEach(i -> checkpointsInZk.add(createHandle(i, id -> createCheckpoint(id, sharedStateRegistry))));
+ store.addCheckpoint(createCheckpoint(1, sharedStateRegistry));
+ store.addCheckpoint(createCheckpoint(5, sharedStateRegistry));
+ store.recover();
+ assertEquals(lastInZk, store.getLatestCheckpoint(false).getCheckpointID());
+ } catch (Exception exception) {
+ throw new RuntimeException(exception);
+ }
+ });
+ }
+
private void testDownloadInternal(TriConsumer<ZooKeeperCompletedCheckpointStore, List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>>, SharedStateRegistry> test) throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
Configuration configuration = new Configuration();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ZooKeeperHighAvailabilityITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ZooKeeperHighAvailabilityITCase.java
deleted file mode 100644
index 81df28f..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ZooKeeperHighAvailabilityITCase.java
+++ /dev/null
@@ -1,456 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.test.checkpointing;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.common.state.ValueStateDescriptor;
-import org.apache.flink.api.common.time.Deadline;
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.client.ClientUtils;
-import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricConfig;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.reporter.MetricReporter;
-import org.apache.flink.runtime.concurrent.FutureUtils;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.metrics.MetricNames;
-import org.apache.flink.runtime.state.CheckpointListener;
-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.testingUtils.TestingUtils;
-import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
-
-import org.apache.curator.test.TestingServer;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.io.StringWriter;
-import java.lang.management.ManagementFactory;
-import java.lang.management.ThreadInfo;
-import java.lang.management.ThreadMXBean;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
-import java.time.Duration;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.hamcrest.core.Is.is;
-import static org.hamcrest.number.OrderingComparison.greaterThan;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Integration tests for {@link org.apache.flink.runtime.checkpoint.ZooKeeperCompletedCheckpointStore}.
- */
-public class ZooKeeperHighAvailabilityITCase extends TestLogger {
-
- private static final Duration TEST_TIMEOUT = Duration.ofSeconds(10000L);
-
- private static final int NUM_JMS = 1;
- private static final int NUM_TMS = 1;
- private static final int NUM_SLOTS_PER_TM = 1;
-
- @ClassRule
- public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
-
- private static File haStorageDir;
-
- private static TestingServer zkServer;
-
- private static MiniClusterWithClientResource miniClusterResource;
-
- private static OneShotLatch waitForCheckpointLatch;
- private static OneShotLatch failInCheckpointLatch;
- private static OneShotLatch blockSnapshotLatch;
-
- @BeforeClass
- public static void setup() throws Exception {
- zkServer = new TestingServer();
-
- Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, NUM_JMS);
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
- config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, NUM_SLOTS_PER_TM);
-
- File haStorageRootDir = TEMPORARY_FOLDER.newFolder();
- String clusterId = UUID.randomUUID().toString();
- haStorageDir = new File(haStorageRootDir, clusterId);
-
- config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, haStorageRootDir.toString());
- config.setString(HighAvailabilityOptions.HA_CLUSTER_ID, clusterId);
- config.setString(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, zkServer.getConnectString());
- config.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");
-
- config.setString(
- ConfigConstants.METRICS_REPORTER_PREFIX + "restarts." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX,
- RestartReporter.class.getName());
-
- // we have to manage this manually because we have to create the ZooKeeper server
- // ahead of this
- miniClusterResource = new MiniClusterWithClientResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(config)
- .setNumberTaskManagers(NUM_TMS)
- .setNumberSlotsPerTaskManager(NUM_SLOTS_PER_TM)
- .build());
-
- miniClusterResource.before();
- }
-
- @AfterClass
- public static void tearDown() throws Exception {
- miniClusterResource.after();
-
- zkServer.stop();
- zkServer.close();
- }
-
- /**
- * Verify that we don't start a job from scratch if we cannot restore any of the
- * CompletedCheckpoints.
- *
- * <p>Synchronization for the different steps and things we want to observe happens via
- * latches in the test method and the methods of {@link CheckpointBlockingFunction}.
- *
- * <p>The test follows these steps:
- * <ol>
- * <li>Start job and block on a latch until we have done some checkpoints
- * <li>Block in the special function
- * <li>Move away the contents of the ZooKeeper HA directory to make restoring from
- * checkpoints impossible
- * <li>Unblock the special function, which now induces a failure
- * <li>Make sure that the job does not recover successfully
- * <li>Move back the HA directory
- * <li>Make sure that the job recovers, we use a latch to ensure that the operator
- * restored successfully
- * </ol>
- */
- @Test
- public void testRestoreBehaviourWithFaultyStateHandles() throws Exception {
- CheckpointBlockingFunction.allowedInitializeCallsWithoutRestore.set(1);
- CheckpointBlockingFunction.successfulRestores.set(0);
- CheckpointBlockingFunction.illegalRestores.set(0);
- CheckpointBlockingFunction.afterMessWithZooKeeper.set(false);
- CheckpointBlockingFunction.failedAlready.set(false);
-
- waitForCheckpointLatch = new OneShotLatch();
- failInCheckpointLatch = new OneShotLatch();
- blockSnapshotLatch = new OneShotLatch();
-
- ClusterClient<?> clusterClient = miniClusterResource.getClusterClient();
- final Deadline deadline = Deadline.now().plus(TEST_TIMEOUT);
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 0));
- env.enableCheckpointing(10); // Flink doesn't allow lower than 10 ms
-
- File checkpointLocation = TEMPORARY_FOLDER.newFolder();
- env.setStateBackend((StateBackend) new FsStateBackend(checkpointLocation.toURI()));
-
- DataStreamSource<String> source = env.addSource(new UnboundedSource());
-
- source
- .keyBy((str) -> str)
- .map(new CheckpointBlockingFunction());
-
- JobGraph jobGraph = env.getStreamGraph().getJobGraph();
- JobID jobID = Preconditions.checkNotNull(jobGraph.getJobID());
-
- ClientUtils.submitJob(clusterClient, jobGraph);
-
- // wait until we did some checkpoints
- waitForCheckpointLatch.await();
-
- log.debug("Messing with HA directory");
- // mess with the HA directory so that the job cannot restore
- File movedCheckpointLocation = TEMPORARY_FOLDER.newFolder();
- AtomicInteger numCheckpoints = new AtomicInteger();
- Files.walkFileTree(haStorageDir.toPath(), new SimpleFileVisitor<Path>() {
- @Override
- public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
- if (file.getFileName().toString().startsWith(ZooKeeperUtils.HA_STORAGE_COMPLETED_CHECKPOINT)) {
- log.debug("Moving original checkpoint file {}.", file);
- try {
- Files.move(file, movedCheckpointLocation.toPath().resolve(file.getFileName()));
- numCheckpoints.incrementAndGet();
- } catch (IOException ioe) {
- // previous checkpoint files may be deleted asynchronously
- log.debug("Exception while moving HA files.", ioe);
- }
- }
- return FileVisitResult.CONTINUE;
- }
- });
-
- // Note to future developers: This will break when we change Flink to not put the
- // checkpoint metadata into the HA directory but instead rely on the fact that the
- // actual checkpoint directory on DFS contains the checkpoint metadata. In this case,
- // ZooKeeper will only contain a "handle" (read: String) that points to the metadata
- // in DFS. The likely solution will be that we have to go directly to ZooKeeper, find
- // out where the checkpoint is stored and mess with that.
- assertTrue(numCheckpoints.get() > 0);
-
- log.debug("Resuming job");
- failInCheckpointLatch.trigger();
-
- assertNotNull("fullRestarts metric could not be accessed.", RestartReporter.numRestarts);
- while (RestartReporter.numRestarts.getValue() < 5 && deadline.hasTimeLeft()) {
- Thread.sleep(50);
- }
- assertThat(RestartReporter.numRestarts.getValue(), is(greaterThan(4L)));
-
- // move back the HA directory so that the job can restore
- CheckpointBlockingFunction.afterMessWithZooKeeper.set(true);
- log.debug("Restored zookeeper");
-
- Files.walkFileTree(movedCheckpointLocation.toPath(), new SimpleFileVisitor<Path>() {
- @Override
- public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
- Files.move(file, haStorageDir.toPath().resolve(file.getFileName()));
- return FileVisitResult.CONTINUE;
- }
- });
- blockSnapshotLatch.trigger();
-
- // now the job should be able to go to RUNNING again and then eventually to FINISHED,
- // which it only does if it could successfully restore
- CompletableFuture<JobStatus> jobStatusFuture = FutureUtils.retrySuccessfulWithDelay(
- () -> clusterClient.getJobStatus(jobID),
- Time.milliseconds(50),
- deadline,
- JobStatus::isGloballyTerminalState,
- TestingUtils.defaultScheduledExecutor());
- try {
- assertEquals(JobStatus.FINISHED, jobStatusFuture.get());
- } catch (Throwable e) {
- // include additional debugging information
- StringWriter error = new StringWriter();
- try (PrintWriter out = new PrintWriter(error)) {
- out.println("The job did not finish in time.");
- out.println("allowedInitializeCallsWithoutRestore= " + CheckpointBlockingFunction.allowedInitializeCallsWithoutRestore.get());
- out.println("illegalRestores= " + CheckpointBlockingFunction.illegalRestores.get());
- out.println("successfulRestores= " + CheckpointBlockingFunction.successfulRestores.get());
- out.println("afterMessWithZooKeeper= " + CheckpointBlockingFunction.afterMessWithZooKeeper.get());
- out.println("failedAlready= " + CheckpointBlockingFunction.failedAlready.get());
- out.println("currentJobStatus= " + clusterClient.getJobStatus(jobID).get());
- out.println("numRestarts= " + RestartReporter.numRestarts.getValue());
- out.println("threadDump= " + generateThreadDump());
- }
- throw new AssertionError(error.toString(), ExceptionUtils.stripCompletionException(e));
- }
-
- assertThat("We saw illegal restores.", CheckpointBlockingFunction.illegalRestores.get(), is(0));
- }
-
- private static String generateThreadDump() {
- final StringBuilder dump = new StringBuilder();
- final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
- final ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 100);
- for (ThreadInfo threadInfo : threadInfos) {
- dump.append('"');
- dump.append(threadInfo.getThreadName());
- dump.append('"');
- final Thread.State state = threadInfo.getThreadState();
- dump.append(System.lineSeparator());
- dump.append(" java.lang.Thread.State: ");
- dump.append(state);
- final StackTraceElement[] stackTraceElements = threadInfo.getStackTrace();
- for (final StackTraceElement stackTraceElement : stackTraceElements) {
- dump.append(System.lineSeparator());
- dump.append(" at ");
- dump.append(stackTraceElement);
- }
- dump.append(System.lineSeparator());
- dump.append(System.lineSeparator());
- }
- return dump.toString();
- }
-
- private static class UnboundedSource implements SourceFunction<String> {
- private volatile boolean running = true;
-
- @Override
- public void run(SourceContext<String> ctx) throws Exception {
- while (running && !CheckpointBlockingFunction.afterMessWithZooKeeper.get()) {
- ctx.collect("hello");
- // don't overdo it ... ;-)
- Thread.sleep(50);
- }
- }
-
- @Override
- public void cancel() {
- running = false;
- }
- }
-
- private static class CheckpointBlockingFunction
- extends RichMapFunction<String, String>
- implements CheckpointedFunction, CheckpointListener {
-
- // verify that we only call initializeState()
- // once with isRestored() == false. All other invocations must have isRestored() == true. This
- // verifies that we don't restart a job from scratch in case the CompletedCheckpoints can't
- // be read.
- static AtomicInteger allowedInitializeCallsWithoutRestore = new AtomicInteger(1);
-
- // we count when we see restores that are not allowed. We only
- // allow restores once we messed with the HA directory and moved it back again
- static AtomicInteger illegalRestores = new AtomicInteger(0);
- static AtomicInteger successfulRestores = new AtomicInteger(0);
-
- // whether we are after the phase where we messed with the ZooKeeper HA directory, i.e.
- // whether it's now ok for a restore to happen
- static AtomicBoolean afterMessWithZooKeeper = new AtomicBoolean(false);
-
- static AtomicBoolean failedAlready = new AtomicBoolean(false);
-
- static AtomicBoolean stateRecorded = new AtomicBoolean(false);
-
- // for checkpoint with a id not less than this id, it includes state data
- static AtomicLong minimalCheckpointIdIncludingData = new AtomicLong(Long.MAX_VALUE);
-
- // make sure there is at least one completed checkpoint including state data
- static AtomicBoolean checkpointCompletedIncludingData = new AtomicBoolean(false);
-
- // also have some state to write to the checkpoint
- private final ValueStateDescriptor<String> stateDescriptor =
- new ValueStateDescriptor<>("state", StringSerializer.INSTANCE);
-
- @Override
- public String map(String value) throws Exception {
- getRuntimeContext().getState(stateDescriptor).update("42");
- stateRecorded.compareAndSet(false, true);
- return value;
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- if (stateRecorded.get()) {
- minimalCheckpointIdIncludingData.compareAndSet(Long.MAX_VALUE,
- context.getCheckpointId());
- }
- if (checkpointCompletedIncludingData.get()) {
- // there is a checkpoint completed with state data, we can trigger the failure now
- waitForCheckpointLatch.trigger();
- failInCheckpointLatch.await();
- if (!failedAlready.getAndSet(true)) {
- throw new RuntimeException("Failing on purpose.");
- } else {
- // make sure there would be no more successful checkpoint before job failing
- // otherwise there might be an unexpected successful checkpoint even
- // CheckpointFailureManager has decided to fail the job
- blockSnapshotLatch.await();
- }
- }
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) {
- if (!context.isRestored()) {
- int updatedValue = allowedInitializeCallsWithoutRestore.decrementAndGet();
- if (updatedValue < 0) {
- illegalRestores.getAndIncrement();
- throw new RuntimeException("We are not allowed any more restores.");
- }
- } else {
- if (!afterMessWithZooKeeper.get()) {
- illegalRestores.getAndIncrement();
- } else if (successfulRestores.getAndIncrement() > 0) {
- // already saw the one allowed successful restore
- illegalRestores.getAndIncrement();
- }
- }
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {
- if (checkpointId >= minimalCheckpointIdIncludingData.get()) {
- checkpointCompletedIncludingData.compareAndSet(false, true);
- }
- }
-
- @Override
- public void notifyCheckpointAborted(long checkpointId) {
- }
- }
-
- /**
- * Reporter that exposes the {@code numRestarts} metric.
- */
- public static class RestartReporter implements MetricReporter {
- static volatile Gauge<Long> numRestarts = null;
-
- @Override
- public void open(MetricConfig metricConfig) {
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public void notifyOfAddedMetric(Metric metric, String name, MetricGroup metricGroup) {
- if (name.equals(MetricNames.NUM_RESTARTS)) {
- numRestarts = (Gauge<Long>) metric;
- }
- }
-
- @Override
- public void notifyOfRemovedMetric(Metric metric, String s, MetricGroup metricGroup) {
- }
- }
-}
[flink] 01/02: [tests][checkpointing] Test retrieval error fails
ZKStore.recover
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4da26b86ee9f57a97dea10ad3607de05f5c9916a
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Oct 21 11:29:33 2020 +0200
[tests][checkpointing] Test retrieval error fails ZKStore.recover
The added test, coupled with the next change (download checkpoints
only if necessary) allows to drop ZK-HA-ITCase. Otherwise,
ZK-HA-ITCase would fail false-positively.
---
.../ZooKeeperCompletedCheckpointStore.java | 5 +-
.../checkpoint/CompletedCheckpointStoreTest.java | 5 +-
.../ZooKeeperCompletedCheckpointStoreTest.java | 78 ++++++++++++++++++++++
3 files changed, 83 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
index b4f72b7..f52d5eb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
@@ -159,6 +159,7 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
// of checkpoints that we don't read due to transient storage outages.
List<CompletedCheckpoint> lastTryRetrievedCheckpoints = new ArrayList<>(numberOfInitialCheckpoints);
List<CompletedCheckpoint> retrievedCheckpoints = new ArrayList<>(numberOfInitialCheckpoints);
+ Exception retrieveException = null;
do {
LOG.info("Trying to fetch {} checkpoints from storage.", numberOfInitialCheckpoints);
@@ -178,6 +179,7 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
}
} catch (Exception e) {
LOG.warn("Could not retrieve checkpoint, not adding to list of recovered checkpoints.", e);
+ retrieveException = e;
}
}
@@ -192,7 +194,8 @@ public class ZooKeeperCompletedCheckpointStore implements CompletedCheckpointSto
if (completedCheckpoints.isEmpty() && numberOfInitialCheckpoints > 0) {
throw new FlinkException(
- "Could not read any of the " + numberOfInitialCheckpoints + " checkpoints from storage.");
+ "Could not read any of the " + numberOfInitialCheckpoints + " checkpoints from storage.",
+ retrieveException);
} else if (completedCheckpoints.size() != numberOfInitialCheckpoints) {
LOG.warn(
"Could only fetch {} of {} checkpoints from storage.",
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
index bdde348..6fe3742 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
@@ -28,7 +28,6 @@ import org.apache.flink.util.TestLogger;
import org.junit.Assert;
import org.junit.Test;
-import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -191,9 +190,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
// ---------------------------------------------------------------------------------------------
- public static TestCompletedCheckpoint createCheckpoint(
- int id,
- SharedStateRegistry sharedStateRegistry) throws IOException {
+ public static TestCompletedCheckpoint createCheckpoint(long id, SharedStateRegistry sharedStateRegistry) {
int numberOfStates = 4;
CheckpointProperties props = CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
index f18c2dd..503dc90 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
@@ -19,14 +19,19 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.runtime.concurrent.Executors;
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
+import org.apache.flink.runtime.state.RetrievableStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.runtime.zookeeper.ZooKeeperResource;
import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.function.TriConsumer;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
@@ -36,8 +41,12 @@ import org.junit.Test;
import javax.annotation.Nonnull;
+import java.util.ArrayList;
import java.util.List;
+import java.util.function.Function;
+import static org.apache.flink.util.ExceptionUtils.findThrowable;
+import static org.apache.flink.util.ExceptionUtils.rethrow;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
@@ -58,6 +67,51 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger {
assertEquals(checkpointId, ZooKeeperCompletedCheckpointStore.pathToCheckpointId(path));
}
+ @Test(expected = ExpectedTestException.class)
+ public void testRecoverFailsIfDownloadFails() throws Exception {
+ testDownloadInternal((store, checkpointsInZk, sharedStateRegistry) -> {
+ try {
+ checkpointsInZk.add(createHandle(1, id -> {
+ throw new ExpectedTestException();
+ }));
+ store.recover();
+ } catch (Exception exception) {
+ findThrowable(exception, ExpectedTestException.class).ifPresent(ExceptionUtils::rethrow);
+ rethrow(exception);
+ }
+ });
+ }
+
+ private void testDownloadInternal(TriConsumer<ZooKeeperCompletedCheckpointStore, List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>>, SharedStateRegistry> test) throws Exception {
+ SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
+ Configuration configuration = new Configuration();
+ configuration.setString(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, zooKeeperResource.getConnectString());
+ List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> checkpointsInZk = new ArrayList<>();
+ ZooKeeperStateHandleStore<CompletedCheckpoint> checkpointsInZooKeeper = new ZooKeeperStateHandleStore<CompletedCheckpoint>(
+ ZooKeeperUtils.startCuratorFramework(configuration),
+ new TestingRetrievableStateStorageHelper<>()) {
+ @Override
+ public List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> getAllAndLock() {
+ return checkpointsInZk;
+ }
+ };
+
+ ZooKeeperCompletedCheckpointStore store = new ZooKeeperCompletedCheckpointStore(10, checkpointsInZooKeeper, Executors.directExecutor());
+ try {
+ test.accept(store, checkpointsInZk, sharedStateRegistry);
+ } finally {
+ store.shutdown(JobStatus.FINISHED);
+ sharedStateRegistry.close();
+ }
+ }
+
+ private Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String> createHandle(long id, Function<Long, CompletedCheckpoint> checkpointSupplier) {
+ return Tuple2.of(
+ new CheckpointStateHandle(checkpointSupplier, id),
+ ZooKeeperCompletedCheckpointStore.checkpointIdToPath(id)
+ );
+ }
+
/**
* Tests that subsumed checkpoints are discarded.
*/
@@ -130,4 +184,28 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger {
Executors.directExecutor());
}
+ private static class CheckpointStateHandle implements RetrievableStateHandle<CompletedCheckpoint> {
+ private static final long serialVersionUID = 1L;
+ private final Function<Long, CompletedCheckpoint> checkpointSupplier;
+ private final long id;
+
+ CheckpointStateHandle(Function<Long, CompletedCheckpoint> checkpointSupplier, long id) {
+ this.checkpointSupplier = checkpointSupplier;
+ this.id = id;
+ }
+
+ @Override
+ public CompletedCheckpoint retrieveState() {
+ return checkpointSupplier.apply(id);
+ }
+
+ @Override
+ public void discardState() {
+ }
+
+ @Override
+ public long getStateSize() {
+ return 0;
+ }
+ }
}