You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2018/12/24 02:24:55 UTC
[flink] branch release-1.7 updated: [FLINK-11179][tests] fix
testCancelSortMatchWhileDoingHeavySorting bug
This is an automated email from the ASF dual-hosted git repository.
jincheng 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 39cbfc6 [FLINK-11179][tests] fix testCancelSortMatchWhileDoingHeavySorting bug
39cbfc6 is described below
commit 39cbfc6303b1c6b83c88b93262f9c3e7449cc33b
Author: sunjincheng121 <su...@gmail.com>
AuthorDate: Mon Dec 17 19:11:57 2018 +0800
[FLINK-11179][tests] fix testCancelSortMatchWhileDoingHeavySorting bug
This closes #7315
---
.../flink/test/cancelling/CancelingTestBase.java | 10 ++--
.../flink/test/cancelling/JoinCancelingITCase.java | 54 ++++++++++++++++++++--
2 files changed, 57 insertions(+), 7 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
index 0d29568..142ca43 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java
@@ -53,6 +53,8 @@ public abstract class CancelingTestBase extends TestLogger {
protected static final int PARALLELISM = 4;
+ protected static final long GET_FUTURE_TIMEOUT = 1000; // 1000 milliseconds
+
// --------------------------------------------------------------------------------------------
@ClassRule
@@ -95,10 +97,10 @@ public abstract class CancelingTestBase extends TestLogger {
Deadline submissionDeadLine = new FiniteDuration(2, TimeUnit.MINUTES).fromNow();
- JobStatus jobStatus = client.getJobStatus(jobSubmissionResult.getJobID()).get(submissionDeadLine.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ JobStatus jobStatus = client.getJobStatus(jobSubmissionResult.getJobID()).get(GET_FUTURE_TIMEOUT, TimeUnit.MILLISECONDS);
while (jobStatus != JobStatus.RUNNING && submissionDeadLine.hasTimeLeft()) {
Thread.sleep(50);
- jobStatus = client.getJobStatus(jobSubmissionResult.getJobID()).get(submissionDeadLine.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ jobStatus = client.getJobStatus(jobSubmissionResult.getJobID()).get(GET_FUTURE_TIMEOUT, TimeUnit.MILLISECONDS);
}
if (jobStatus != JobStatus.RUNNING) {
Assert.fail("Job not in state RUNNING.");
@@ -110,10 +112,10 @@ public abstract class CancelingTestBase extends TestLogger {
Deadline cancelDeadline = new FiniteDuration(maxTimeTillCanceled, TimeUnit.MILLISECONDS).fromNow();
- JobStatus jobStatusAfterCancel = client.getJobStatus(jobSubmissionResult.getJobID()).get(cancelDeadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ JobStatus jobStatusAfterCancel = client.getJobStatus(jobSubmissionResult.getJobID()).get(GET_FUTURE_TIMEOUT, TimeUnit.MILLISECONDS);
while (jobStatusAfterCancel != JobStatus.CANCELED && cancelDeadline.hasTimeLeft()) {
Thread.sleep(50);
- jobStatusAfterCancel = client.getJobStatus(jobSubmissionResult.getJobID()).get(cancelDeadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
+ jobStatusAfterCancel = client.getJobStatus(jobSubmissionResult.getJobID()).get(GET_FUTURE_TIMEOUT, TimeUnit.MILLISECONDS);
}
if (jobStatusAfterCancel != JobStatus.CANCELED) {
Assert.fail("Failed to cancel job with ID " + jobSubmissionResult.getJobID() + '.');
diff --git a/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
index 7288ba5..9a6ae44 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/cancelling/JoinCancelingITCase.java
@@ -20,19 +20,21 @@ package org.apache.flink.test.cancelling;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.RichJoinFunction;
+import org.apache.flink.api.common.io.GenericInputFormat;
import org.apache.flink.api.common.operators.base.JoinOperatorBase;
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.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.operators.testutils.UniformIntTupleGenerator;
import org.apache.flink.test.util.InfiniteIntegerTupleInputFormat;
import org.apache.flink.test.util.UniformIntTupleGeneratorInputFormat;
import org.junit.Ignore;
import org.junit.Test;
+import java.io.IOException;
+
/**
* Test job cancellation from within a JoinFunction.
*/
@@ -78,7 +80,6 @@ public class JoinCancelingITCase extends CancelingTestBase {
private void executeTaskWithGenerator(
JoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> joiner,
int keys, int vals, int msecsTillCanceling, int maxTimeTillCanceled) throws Exception {
- UniformIntTupleGenerator g = new UniformIntTupleGenerator(keys, vals, false);
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Tuple2<Integer, Integer>> input1 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
DataSet<Tuple2<Integer, Integer>> input2 = env.createInput(new UniformIntTupleGeneratorInputFormat(keys, vals));
@@ -96,7 +97,25 @@ public class JoinCancelingITCase extends CancelingTestBase {
@Test
public void testCancelSortMatchWhileDoingHeavySorting() throws Exception {
- executeTaskWithGenerator(new SimpleMatcher<Integer>(), 50000, 100, 30 * 1000, 30 * 1000);
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ HeavyCompareGeneratorInputFormat input = new HeavyCompareGeneratorInputFormat(100);
+ DataSet<Tuple2<HeavyCompare, Integer>> input1 = env.createInput(input);
+ DataSet<Tuple2<HeavyCompare, Integer>> input2 = env.createInput(input);
+
+ input1.join(input2, JoinOperatorBase.JoinHint.REPARTITION_SORT_MERGE)
+ .where(0)
+ .equalTo(0)
+ .with(new JoinFunction<Tuple2<HeavyCompare, Integer>, Tuple2<HeavyCompare, Integer>, Tuple2<HeavyCompare, Integer>>() {
+ @Override
+ public Tuple2<HeavyCompare, Integer> join(
+ Tuple2<HeavyCompare, Integer> first,
+ Tuple2<HeavyCompare, Integer> second) throws Exception {
+ throw new Exception("Job should be canceled in sort-merge phase, never run here ...");
+ }
+ })
+ .output(new DiscardingOutputFormat<Tuple2<HeavyCompare, Integer>>());
+
+ runAndCancelJob(env.createProgramPlan(), 30 * 1000, 60 * 1000);
}
// --------------- Test Sort Matches that are canceled while in the Matching Phase -----------------
@@ -175,3 +194,32 @@ public class JoinCancelingITCase extends CancelingTestBase {
}
}
}
+
+class HeavyCompare implements Comparable<HeavyCompare>, java.io.Serializable {
+ @Override
+ public int compareTo(org.apache.flink.test.cancelling.HeavyCompare o) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException iex) {}
+ return 0;
+ }
+}
+
+class HeavyCompareGeneratorInputFormat extends GenericInputFormat<Tuple2<HeavyCompare, Integer>> {
+ private int valueTotal;
+
+ public HeavyCompareGeneratorInputFormat(int numVals) {
+ this.valueTotal = numVals;
+ }
+
+ @Override
+ public boolean reachedEnd() throws IOException {
+ return valueTotal <= 0;
+ }
+
+ @Override
+ public Tuple2<HeavyCompare, Integer> nextRecord(Tuple2<HeavyCompare, Integer> reuse) throws IOException {
+ valueTotal -= 1;
+ return new Tuple2<>(new HeavyCompare(), 20110701);
+ }
+}