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 2018/03/26 07:33:34 UTC
[4/5] flink git commit: [FLINK-8721][flip6] Handle archiving failures
for accumulators
http://git-wip-us.apache.org/repos/asf/flink/blob/a6a7623d/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
index 282b192..3d90833 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorErrorITCase.java
@@ -18,25 +18,24 @@
package org.apache.flink.test.accumulators;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.DoubleCounter;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.DiscardingOutputFormat;
-import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
-import org.apache.flink.test.util.TestEnvironment;
+import org.apache.flink.test.util.MiniClusterResource;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TestLogger;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -45,84 +44,80 @@ import static org.junit.Assert.fail;
* b) are not compatible with existing accumulator.
*/
public class AccumulatorErrorITCase extends TestLogger {
-
- private static LocalFlinkMiniCluster cluster;
-
- private static ExecutionEnvironment env;
-
- @BeforeClass
- public static void startCluster() {
+ private static final String FAULTY_CLONE_ACCUMULATOR = "faulty-clone";
+ private static final String FAULTY_MERGE_ACCUMULATOR = "faulty-merge";
+ private static final String INCOMPATIBLE_ACCUMULATORS_NAME = "incompatible-accumulators";
+
+ @ClassRule
+ public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ new MiniClusterResource.MiniClusterResourceConfiguration(
+ getConfiguration(),
+ 2,
+ 3));
+
+ public static Configuration getConfiguration() {
Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 3);
config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 12L);
- cluster = new LocalFlinkMiniCluster(config, false);
-
- cluster.start();
-
- env = new TestEnvironment(cluster, 6, false);
- }
-
- @AfterClass
- public static void shutdownCluster() {
- cluster.stop();
- cluster = null;
+ return config;
}
@Test
public void testFaultyAccumulator() throws Exception {
-
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
// Test Exception forwarding with faulty Accumulator implementation
- DataSet<Long> input = env.generateSequence(0, 10000);
+ env.generateSequence(0, 10000)
+ .map(new FaultyAccumulatorUsingMapper())
+ .output(new DiscardingOutputFormat<>());
- DataSet<Long> map = input.map(new FaultyAccumulatorUsingMapper());
-
- map.output(new DiscardingOutputFormat<Long>());
-
- try {
- env.execute();
- fail("Should have failed.");
- } catch (JobExecutionException e) {
- Assert.assertTrue("Root cause should be:",
- e.getCause() instanceof CustomException);
- }
+ assertAccumulatorsShouldFail(env.execute());
}
@Test
public void testInvalidTypeAccumulator() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
// Test Exception forwarding with faulty Accumulator implementation
- DataSet<Long> input = env.generateSequence(0, 10000);
-
- DataSet<Long> mappers = input.map(new IncompatibleAccumulatorTypesMapper())
- .map(new IncompatibleAccumulatorTypesMapper2());
-
- mappers.output(new DiscardingOutputFormat<Long>());
+ env.generateSequence(0, 10000)
+ .map(new IncompatibleAccumulatorTypesMapper())
+ .map(new IncompatibleAccumulatorTypesMapper2())
+ .output(new DiscardingOutputFormat<>());
try {
env.execute();
fail("Should have failed.");
} catch (JobExecutionException e) {
- Assert.assertTrue("Root cause should be:",
+ assertTrue("Root cause should be:",
e.getCause() instanceof Exception);
- Assert.assertTrue("Root cause should be:",
+ assertTrue("Root cause should be:",
e.getCause().getCause() instanceof UnsupportedOperationException);
}
}
+ @Test
+ public void testFaultyMergeAccumulator() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().disableSysoutLogging();
+
+ // Test Exception forwarding with faulty Accumulator implementation
+ env.generateSequence(0, 10000)
+ .map(new FaultyMergeAccumulatorUsingMapper())
+ .output(new DiscardingOutputFormat<>());
+
+ assertAccumulatorsShouldFail(env.execute());
+ }
+
/* testFaultyAccumulator */
private static class FaultyAccumulatorUsingMapper extends RichMapFunction<Long, Long> {
-
private static final long serialVersionUID = 42;
@Override
public void open(Configuration parameters) throws Exception {
- getRuntimeContext().addAccumulator("test", new FaultyAccumulator());
+ getRuntimeContext().addAccumulator(FAULTY_CLONE_ACCUMULATOR, new FaultyCloneAccumulator());
}
@Override
@@ -131,8 +126,7 @@ public class AccumulatorErrorITCase extends TestLogger {
}
}
- private static class FaultyAccumulator extends LongCounter {
-
+ private static class FaultyCloneAccumulator extends LongCounter {
private static final long serialVersionUID = 42;
@Override
@@ -141,19 +135,14 @@ public class AccumulatorErrorITCase extends TestLogger {
}
}
- private static class CustomException extends RuntimeException {
- private static final long serialVersionUID = 42;
- }
-
/* testInvalidTypeAccumulator */
private static class IncompatibleAccumulatorTypesMapper extends RichMapFunction<Long, Long> {
-
private static final long serialVersionUID = 42;
@Override
public void open(Configuration parameters) throws Exception {
- getRuntimeContext().addAccumulator("test", new LongCounter());
+ getRuntimeContext().addAccumulator(INCOMPATIBLE_ACCUMULATORS_NAME, new LongCounter());
}
@Override
@@ -163,12 +152,27 @@ public class AccumulatorErrorITCase extends TestLogger {
}
private static class IncompatibleAccumulatorTypesMapper2 extends RichMapFunction<Long, Long> {
+ private static final long serialVersionUID = 42;
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ getRuntimeContext().addAccumulator(INCOMPATIBLE_ACCUMULATORS_NAME, new DoubleCounter());
+ }
+
+ @Override
+ public Long map(Long value) throws Exception {
+ return -1L;
+ }
+ }
+
+ /** */
+ private static class FaultyMergeAccumulatorUsingMapper extends RichMapFunction<Long, Long> {
private static final long serialVersionUID = 42;
@Override
public void open(Configuration parameters) throws Exception {
- getRuntimeContext().addAccumulator("test", new DoubleCounter());
+ getRuntimeContext().addAccumulator(FAULTY_MERGE_ACCUMULATOR, new FaultyMergeAccumulator());
}
@Override
@@ -177,4 +181,31 @@ public class AccumulatorErrorITCase extends TestLogger {
}
}
+ private static class FaultyMergeAccumulator extends LongCounter {
+ private static final long serialVersionUID = 42;
+
+ @Override
+ public void merge(Accumulator<Long, Long> other) {
+ throw new CustomException();
+ }
+
+ @Override
+ public LongCounter clone() {
+ return new FaultyMergeAccumulator();
+ }
+ }
+
+ private static class CustomException extends RuntimeException {
+ private static final long serialVersionUID = 42;
+ }
+
+ private static void assertAccumulatorsShouldFail(JobExecutionResult result) {
+ try {
+ result.getAllAccumulatorResults();
+ fail("Should have failed");
+ }
+ catch (Exception ex) {
+ assertTrue(ExceptionUtils.findThrowable(ex, CustomException.class).isPresent());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a6a7623d/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
index b7f54fa..9f87a56 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
@@ -76,6 +76,7 @@ public class AccumulatorITCase extends JavaProgramTestBase {
System.out.println(AccumulatorHelper.getResultsFormatted(res.getAllAccumulatorResults()));
Assert.assertEquals(Integer.valueOf(3), res.getAccumulatorResult("num-lines"));
+ Assert.assertEquals(Integer.valueOf(3), res.getIntCounterResult("num-lines"));
Assert.assertEquals(Double.valueOf(getParallelism()), res.getAccumulatorResult("open-close-counter"));
http://git-wip-us.apache.org/repos/asf/flink/blob/a6a7623d/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
index ff362dd..4a2219a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
@@ -160,7 +160,7 @@ public class AccumulatorLiveITCase extends TestLogger {
deadline,
accumulators -> accumulators.size() == 1
&& accumulators.containsKey(ACCUMULATOR_NAME)
- && (int) accumulators.get(ACCUMULATOR_NAME) == NUM_ITERATIONS,
+ && (int) accumulators.get(ACCUMULATOR_NAME).getUnchecked() == NUM_ITERATIONS,
TestingUtils.defaultScheduledExecutor()
).get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
http://git-wip-us.apache.org/repos/asf/flink/blob/a6a7623d/flink-tests/src/test/java/org/apache/flink/test/accumulators/LegacyAccumulatorLiveITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/LegacyAccumulatorLiveITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/LegacyAccumulatorLiveITCase.java
index 6595b10..b273ade 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/LegacyAccumulatorLiveITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/LegacyAccumulatorLiveITCase.java
@@ -49,6 +49,7 @@ import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.TestLogger;
import akka.actor.ActorRef;
@@ -195,7 +196,7 @@ public class LegacyAccumulatorLiveITCase extends TestLogger {
expectMsgClass(TIMEOUT, JobManagerMessages.JobSubmitSuccess.class);
TestingJobManagerMessages.UpdatedAccumulators msg = (TestingJobManagerMessages.UpdatedAccumulators) receiveOne(TIMEOUT);
- Map<String, Accumulator<?, ?>> userAccumulators = msg.userAccumulators();
+ Map<String, OptionalFailure<Accumulator<?, ?>>> userAccumulators = msg.userAccumulators();
ExecutionAttemptID mapperTaskID = null;
@@ -264,9 +265,9 @@ public class LegacyAccumulatorLiveITCase extends TestLogger {
}};
}
- private static boolean checkUserAccumulators(int expected, Map<String, Accumulator<?, ?>> accumulatorMap) {
+ private static boolean checkUserAccumulators(int expected, Map<String, OptionalFailure<Accumulator<?, ?>>> accumulatorMap) {
LOG.info("checking user accumulators");
- return accumulatorMap.containsKey(ACCUMULATOR_NAME) && expected == ((IntCounter) accumulatorMap.get(ACCUMULATOR_NAME)).getLocalValue();
+ return accumulatorMap.containsKey(ACCUMULATOR_NAME) && expected == ((IntCounter) accumulatorMap.get(ACCUMULATOR_NAME).getUnchecked()).getLocalValue();
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/a6a7623d/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
index 91b5de8..8f2aaa1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
@@ -32,6 +32,7 @@ import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.util.MiniClusterResource;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.OptionalFailure;
import org.apache.commons.io.FileUtils;
import org.junit.BeforeClass;
@@ -138,11 +139,11 @@ public abstract class SavepointMigrationTestBase extends TestBaseUtils {
boolean done = false;
while (DEADLINE.hasTimeLeft()) {
Thread.sleep(100);
- Map<String, Object> accumulators = client.getAccumulators(jobSubmissionResult.getJobID());
+ Map<String, OptionalFailure<Object>> accumulators = client.getAccumulators(jobSubmissionResult.getJobID());
boolean allDone = true;
for (Tuple2<String, Integer> acc : expectedAccumulators) {
- Integer numFinished = (Integer) accumulators.get(acc.f0);
+ Integer numFinished = (Integer) accumulators.get(acc.f0).get();
if (numFinished == null) {
allDone = false;
break;
@@ -211,16 +212,16 @@ public abstract class SavepointMigrationTestBase extends TestBaseUtils {
}
Thread.sleep(100);
- Map<String, Object> accumulators = client.getAccumulators(jobId);
+ Map<String, OptionalFailure<Object>> accumulators = client.getAccumulators(jobId);
boolean allDone = true;
for (Tuple2<String, Integer> acc : expectedAccumulators) {
- Integer numFinished = (Integer) accumulators.get(acc.f0);
+ OptionalFailure<Object> numFinished = accumulators.get(acc.f0);
if (numFinished == null) {
allDone = false;
break;
}
- if (!numFinished.equals(acc.f1)) {
+ if (!numFinished.get().equals(acc.f1)) {
allDone = false;
break;
}