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/11/05 11:33:19 UTC
[flink] branch release-1.7 updated: [FLINK-10773] Harden resume
externalized checkpoint end-to-end test
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.7 by this push:
new 80bed3d [FLINK-10773] Harden resume externalized checkpoint end-to-end test
80bed3d is described below
commit 80bed3d87b2722f0f6c04930b3ab1bdde860962c
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Sun Nov 4 19:32:44 2018 +0100
[FLINK-10773] Harden resume externalized checkpoint end-to-end test
Ignore the 'Artificial Failure' exceptions and rename
ExceptionThrowingFailureMapper into FailureMapper to avoid false
positive exception matchings.
---
.../flink/streaming/tests/DataStreamAllroundTestJobFactory.java | 6 ++++--
.../apache/flink/streaming/tests/DataStreamAllroundTestProgram.java | 6 +++---
.../{ExceptionThrowingFailureMapper.java => FailureMapper.java} | 4 ++--
flink-end-to-end-tests/test-scripts/common.sh | 1 +
4 files changed, 10 insertions(+), 7 deletions(-)
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
index fb92960..3c8d0ad 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestJobFactory.java
@@ -322,6 +322,8 @@ class DataStreamAllroundTestJobFactory {
SEQUENCE_GENERATOR_SRC_EVENT_TIME_MAX_OUT_OF_ORDERNESS.key(),
SEQUENCE_GENERATOR_SRC_EVENT_TIME_MAX_OUT_OF_ORDERNESS.defaultValue()))) {
+ private static final long serialVersionUID = -3154419724891779938L;
+
@Override
public long extractTimestamp(Event element) {
return element.getEventTime();
@@ -367,8 +369,8 @@ class DataStreamAllroundTestJobFactory {
return pt.getBoolean(TEST_SIMULATE_FAILURE.key(), TEST_SIMULATE_FAILURE.defaultValue());
}
- static MapFunction<Event, Event> createExceptionThrowingFailureMapper(ParameterTool pt) {
- return new ExceptionThrowingFailureMapper<>(
+ static MapFunction<Event, Event> createFailureMapper(ParameterTool pt) {
+ return new FailureMapper<>(
pt.getLong(
TEST_SIMULATE_FAILURE_NUM_RECORDS.key(),
TEST_SIMULATE_FAILURE_NUM_RECORDS.defaultValue()),
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
index 70fdade..b14e2af 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
@@ -41,7 +41,7 @@ import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createArtificialKeyedStateMapper;
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createArtificialOperatorStateMapper;
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createEventSource;
-import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createExceptionThrowingFailureMapper;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createFailureMapper;
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createSemanticsCheckMapper;
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createTimestampExtractor;
import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.isSimulateFailures;
@@ -68,7 +68,7 @@ public class DataStreamAllroundTestProgram {
private static final String OPERATOR_STATE_OPER_NAME = "ArtificalOperatorStateMapper";
private static final String TIME_WINDOW_OPER_NAME = "TumblingWindowOperator";
private static final String SEMANTICS_CHECK_MAPPER_NAME = "SemanticsCheckMapper";
- private static final String FAILURE_MAPPER_NAME = "ExceptionThrowingFailureMapper";
+ private static final String FAILURE_MAPPER_NAME = "FailureMapper";
public static void main(String[] args) throws Exception {
final ParameterTool pt = ParameterTool.fromArgs(args);
@@ -145,7 +145,7 @@ public class DataStreamAllroundTestProgram {
if (isSimulateFailures(pt)) {
eventStream3 = eventStream3
- .map(createExceptionThrowingFailureMapper(pt))
+ .map(createFailureMapper(pt))
.setParallelism(1)
.name(FAILURE_MAPPER_NAME);
}
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/ExceptionThrowingFailureMapper.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/FailureMapper.java
similarity index 94%
rename from flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/ExceptionThrowingFailureMapper.java
rename to flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/FailureMapper.java
index d758ef5..a3a1c25 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/ExceptionThrowingFailureMapper.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/FailureMapper.java
@@ -30,7 +30,7 @@ import org.apache.flink.runtime.state.CheckpointListener;
* of the operator can also be configured. Note that this also takes into account
* failures that were not triggered by this mapper, e.g. TaskManager failures.
*/
-public class ExceptionThrowingFailureMapper<T> extends RichMapFunction<T, T> implements CheckpointListener {
+public class FailureMapper<T> extends RichMapFunction<T, T> implements CheckpointListener {
private static final long serialVersionUID = -5286927943454740016L;
@@ -41,7 +41,7 @@ public class ExceptionThrowingFailureMapper<T> extends RichMapFunction<T, T> imp
private long numProcessedRecords;
private long numCompleteCheckpoints;
- public ExceptionThrowingFailureMapper(
+ public FailureMapper(
long numProcessedRecordsFailureThreshold,
long numCompleteCheckpointsFailureThreshold,
int maxNumFailures) {
diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 09c42af..bdf6f64 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -347,6 +347,7 @@ function check_logs_for_exceptions {
| grep -v "java.lang.Exception: Execution was suspended" \
| grep -v "java.io.InvalidClassException: org.apache.flink.formats.avro.typeutils.AvroSerializer" \
| grep -v "Caused by: java.lang.Exception: JobManager is shutting down" \
+ | grep -v "java.lang.Exception: Artificial failure" \
| grep -ic "exception")
if [[ ${exception_count} -gt 0 ]]; then
echo "Found exception in log files:"