You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/03/21 22:39:43 UTC
[1/8] flink git commit: [FLINK-8984][network] Drop
taskmanager.exactly-once.blocking.data.enabled config option
Repository: flink
Updated Branches:
refs/heads/release-1.5 9f2b3c332 -> f0e6ae6eb
[FLINK-8984][network] Drop taskmanager.exactly-once.blocking.data.enabled config option
Previously there were twe options:
taskmanager.network.credit-based-flow-control.enabled
and
taskmanager.exactly-once.blocking.data.enabled
If we disabled first one, but keept default value for the second one deadlocks will occur.
By dropping taskmanager.exactly-once.blocking.data.enabled we can always use:
- blocking BarrierBuffer for credit based flow control
- spilling BarrierBuffer for non credit based flow control.
This closes #5708.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/23005ee1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/23005ee1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/23005ee1
Branch: refs/heads/release-1.5
Commit: 23005ee1f653a3fcddac0f710d56f3e0d6157a48
Parents: 0fa76e5
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Fri Mar 16 13:28:08 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:01:31 2018 +0100
----------------------------------------------------------------------
.../apache/flink/configuration/TaskManagerOptions.java | 12 ------------
.../flink/streaming/runtime/io/InputProcessorUtil.java | 2 +-
2 files changed, 1 insertion(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/23005ee1/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index 4e08fda..c7b0782 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -326,18 +326,6 @@ public class TaskManagerOptions {
.defaultValue(true)
.withDescription("Boolean flag to enable/disable network credit-based flow control.");
- /**
- * Config parameter defining whether to spill data for channels with barrier or not in exactly-once
- * mode based on credit-based flow control.
- *
- * @deprecated Will be removed for Flink 1.6 when the old code will be dropped in favour of
- * credit-based flow control.
- */
- @Deprecated
- public static final ConfigOption<Boolean> EXACTLY_ONCE_BLOCKING_DATA_ENABLED =
- key("taskmanager.exactly-once.blocking.data.enabled")
- .defaultValue(true);
-
// ------------------------------------------------------------------------
// Task Options
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/23005ee1/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
index cb56eee..1ae34b3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
@@ -51,7 +51,7 @@ public class InputProcessorUtil {
+ " must be positive or -1 (infinite)");
}
- if (taskManagerConfig.getBoolean(TaskManagerOptions.EXACTLY_ONCE_BLOCKING_DATA_ENABLED)) {
+ if (taskManagerConfig.getBoolean(TaskManagerOptions.NETWORK_CREDIT_BASED_FLOW_CONTROL_ENABLED)) {
barrierHandler = new BarrierBuffer(inputGate, new CachedBufferBlocker(inputGate.getPageSize()), maxAlign);
} else {
barrierHandler = new BarrierBuffer(inputGate, new BufferSpiller(ioManager, inputGate.getPageSize()), maxAlign);
[4/8] flink git commit: [FLINK-8073][kafka-tests] Disable timeout in
tests
Posted by ch...@apache.org.
[FLINK-8073][kafka-tests] Disable timeout in tests
To get stacktraces in case of deadlock do not timeout tests programatically.
This closes #5718.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2b85b464
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2b85b464
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2b85b464
Branch: refs/heads/release-1.5
Commit: 2b85b464d52942541b5c6d63fddd70a18a36ab9e
Parents: 9f2b3c3
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Mon Mar 19 09:48:15 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:01:31 2018 +0100
----------------------------------------------------------------------
.../connectors/kafka/FlinkKafkaProducer011ITCase.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2b85b464/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
index 81bf0bf..361f269 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
@@ -80,7 +80,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
* This test ensures that transactions reusing transactional.ids (after returning to the pool) will not clash
* with previous transactions using same transactional.ids.
*/
- @Test(timeout = 120_000L)
+ @Test
public void testRestoreToCheckpointAfterExceedingProducersPool() throws Exception {
String topic = "flink-kafka-producer-fail-before-notify";
@@ -123,7 +123,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
}
}
- @Test(timeout = 120_000L)
+ @Test
public void testFlinkKafkaProducer011FailBeforeNotify() throws Exception {
String topic = "flink-kafka-producer-fail-before-notify";
@@ -166,7 +166,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
deleteTestTopic(topic);
}
- @Test(timeout = 120_000L)
+ @Test
public void testFlinkKafkaProducer011FailTransactionCoordinatorBeforeNotify() throws Exception {
String topic = "flink-kafka-producer-fail-transaction-coordinator-before-notify";
@@ -221,7 +221,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
* If such transactions were left alone lingering it consumers would be unable to read committed records
* that were created after this lingering transaction.
*/
- @Test(timeout = 120_000L)
+ @Test
public void testFailBeforeNotifyAndResumeWorkAfterwards() throws Exception {
String topic = "flink-kafka-producer-fail-before-notify";
@@ -263,7 +263,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
deleteTestTopic(topic);
}
- @Test(timeout = 120_000L)
+ @Test
public void testFailAndRecoverSameCheckpointTwice() throws Exception {
String topic = "flink-kafka-producer-fail-and-recover-same-checkpoint-twice";
@@ -316,7 +316,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
* If such transactions were left alone lingering it consumers would be unable to read committed records
* that were created after this lingering transaction.
*/
- @Test(timeout = 120_000L)
+ @Test
public void testScaleDownBeforeFirstCheckpoint() throws Exception {
String topic = "scale-down-before-first-checkpoint";
@@ -381,7 +381,7 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
* new subtask have to generate new id(s), but he can not use ids that are potentially in use, so it has to generate
* new ones that are greater then 4.
*/
- @Test(timeout = 120_000L)
+ @Test
public void testScaleUpAfterScalingDown() throws Exception {
String topic = "scale-down-before-first-checkpoint";
[3/8] flink git commit: [hotfix][kafka-tests] Clean up and drop
unused field in KafkaProducerTestBase
Posted by ch...@apache.org.
[hotfix][kafka-tests] Clean up and drop unused field in KafkaProducerTestBase
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/29fd6ce3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/29fd6ce3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/29fd6ce3
Branch: refs/heads/release-1.5
Commit: 29fd6ce316d26c85647cc2b341fef20d983b420f
Parents: 2b85b46
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Tue Mar 20 11:17:18 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:01:31 2018 +0100
----------------------------------------------------------------------
.../connectors/kafka/KafkaProducerTestBase.java | 13 ++++---------
1 file changed, 4 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/29fd6ce3/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 8104d8f..9278b67 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -482,8 +482,7 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
private static final long serialVersionUID = 6334389850158707313L;
- public static volatile boolean restartedLeaderBefore;
- public static volatile boolean hasBeenCheckpointedBeforeFailure;
+ public static volatile boolean triggeredShutdown;
public static volatile int numElementsBeforeSnapshot;
public static volatile Runnable shutdownAction;
@@ -491,11 +490,9 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
private int numElementsTotal;
private boolean failer;
- private boolean hasBeenCheckpointed;
public static void resetState(Runnable shutdownAction) {
- restartedLeaderBefore = false;
- hasBeenCheckpointedBeforeFailure = false;
+ triggeredShutdown = false;
numElementsBeforeSnapshot = 0;
BrokerRestartingMapper.shutdownAction = shutdownAction;
}
@@ -513,13 +510,12 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
public T map(T value) throws Exception {
numElementsTotal++;
- if (!restartedLeaderBefore) {
+ if (!triggeredShutdown) {
Thread.sleep(10);
if (failer && numElementsTotal >= failCount) {
// shut down a Kafka broker
- hasBeenCheckpointedBeforeFailure = hasBeenCheckpointed;
- restartedLeaderBefore = true;
+ triggeredShutdown = true;
shutdownAction.run();
}
}
@@ -528,7 +524,6 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
@Override
public void notifyCheckpointComplete(long checkpointId) {
- hasBeenCheckpointed = true;
}
@Override
[6/8] flink git commit: [FLINK-8649] [scala api] Pass on TypeInfo in
StreamExecutionEnvironment.createInput
Posted by ch...@apache.org.
[FLINK-8649] [scala api] Pass on TypeInfo in StreamExecutionEnvironment.createInput
This closes #5478.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0e2bf7ee
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0e2bf7ee
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0e2bf7ee
Branch: refs/heads/release-1.5
Commit: 0e2bf7eea9e2b1d694711a33662b0d5f64d0e10f
Parents: 16b9097
Author: Gabor Gevay <gg...@gmail.com>
Authored: Tue Feb 13 18:09:05 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:04:12 2018 +0100
----------------------------------------------------------------------
.../streaming/api/scala/StreamExecutionEnvironment.scala | 7 ++++++-
.../org/apache/flink/streaming/api/scala/DataStreamTest.scala | 7 +++++++
2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0e2bf7ee/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index cd96dbf..9410a95 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.{Internal, Public, PublicEvolving}
import org.apache.flink.api.common.io.{FileInputFormat, FilePathFilter, InputFormat}
import org.apache.flink.api.common.restartstrategy.RestartStrategies.RestartStrategyConfiguration
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.configuration.Configuration
@@ -594,7 +595,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
*/
@PublicEvolving
def createInput[T: TypeInformation](inputFormat: InputFormat[T, _]): DataStream[T] =
- asScalaStream(javaEnv.createInput(inputFormat))
+ if (inputFormat.isInstanceOf[ResultTypeQueryable[_]]) {
+ asScalaStream(javaEnv.createInput(inputFormat))
+ } else {
+ asScalaStream(javaEnv.createInput(inputFormat, implicitly[TypeInformation[T]]))
+ }
/**
* Create a DataStream using a user defined source function for arbitrary
http://git-wip-us.apache.org/repos/asf/flink/blob/0e2bf7ee/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala
index 51ec5e3..9e1c493 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.api.scala
import java.lang
import org.apache.flink.api.common.functions._
+import org.apache.flink.api.java.io.ParallelIteratorInputFormat
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
@@ -673,6 +674,12 @@ class DataStreamTest extends AbstractTestBase {
assert(sg.getIterationSourceSinkPairs.size() == 2)
}
+ @Test
+ def testCreateInputPassesOnTypeInfo(): Unit = {
+ StreamExecutionEnvironment.getExecutionEnvironment.createInput[Tuple1[Integer]](
+ new ParallelIteratorInputFormat[Tuple1[Integer]](null))
+ }
+
/////////////////////////////////////////////////////////////
// Utilities
/////////////////////////////////////////////////////////////
[7/8] flink git commit: Revert "[FLINK-8703][tests] Port
AutoParallelismITCase to flip6"
Posted by ch...@apache.org.
Revert "[FLINK-8703][tests] Port AutoParallelismITCase to flip6"
The test does not actually run on Flip6, see FLINK-8813.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c12e21cc
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c12e21cc
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c12e21cc
Branch: refs/heads/release-1.5
Commit: c12e21cc8cb8b809689cc35c66e5363723352e22
Parents: 0e2bf7e
Author: zentol <ch...@apache.org>
Authored: Wed Mar 21 20:59:05 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:04:15 2018 +0100
----------------------------------------------------------------------
.../flink/test/misc/AutoParallelismITCase.java | 42 +++++++++++++++-----
1 file changed, 32 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c12e21cc/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java b/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
index c25dbf0..9cafee6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/AutoParallelismITCase.java
@@ -22,15 +22,17 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.RichMapPartitionFunction;
import org.apache.flink.api.common.io.GenericInputFormat;
import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.GenericInputSplit;
-import org.apache.flink.test.util.MiniClusterResource;
+import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.test.util.TestEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
import org.junit.Test;
import java.io.IOException;
@@ -50,17 +52,37 @@ public class AutoParallelismITCase extends TestLogger {
private static final int SLOTS_PER_TM = 7;
private static final int PARALLELISM = NUM_TM * SLOTS_PER_TM;
- @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
- new MiniClusterResource.MiniClusterResourceConfiguration(
- new Configuration(),
- 2,
- 7));
+ private static LocalFlinkMiniCluster cluster;
+
+ private static TestEnvironment env;
+
+ @BeforeClass
+ public static void setupCluster() {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TM);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, SLOTS_PER_TM);
+ cluster = new LocalFlinkMiniCluster(config, false);
+
+ cluster.start();
+
+ env = new TestEnvironment(cluster, NUM_TM * SLOTS_PER_TM, false);
+ }
+
+ @AfterClass
+ public static void teardownCluster() {
+ try {
+ cluster.stop();
+ }
+ catch (Throwable t) {
+ System.err.println("Error stopping cluster on shutdown");
+ t.printStackTrace();
+ fail("ClusterClient shutdown caused an exception: " + t.getMessage());
+ }
+ }
@Test
public void testProgramWithAutoParallelism() {
try {
- ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
env.getConfig().disableSysoutLogging();
[5/8] flink git commit: [FLINK-9020][E2ETests] Use separate modules
per testcase
Posted by ch...@apache.org.
[FLINK-9020][E2ETests] Use separate modules per testcase
This closes #5717.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/16b90976
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/16b90976
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/16b90976
Branch: refs/heads/release-1.5
Commit: 16b90976cb05dcb5683c3bfed34ff3ec04e7416d
Parents: 23005ee
Author: Florian Schmidt <fl...@icloud.com>
Authored: Mon Mar 19 13:26:14 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:04:04 2018 +0100
----------------------------------------------------------------------
.../pom.xml | 108 +++++++++++++++++++
.../flink/runtime/taskmanager/TaskManager.java | 30 ++++++
.../streaming/tests/ClassLoaderTestProgram.java | 102 ++++++++++++++++++
.../src/main/resources/.version.properties | 1 +
flink-end-to-end-tests/pom.xml | 78 ++------------
.../flink/runtime/taskmanager/TaskManager.java | 30 ------
.../streaming/tests/ClassLoaderTestProgram.java | 102 ------------------
.../src/main/resources/.version.properties | 1 -
.../test-scripts/test_streaming_classloader.sh | 4 +-
9 files changed, 249 insertions(+), 207 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/flink-parent-child-classloading-test/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-parent-child-classloading-test/pom.xml b/flink-end-to-end-tests/flink-parent-child-classloading-test/pom.xml
new file mode 100644
index 0000000..a7d09c0
--- /dev/null
+++ b/flink-end-to-end-tests/flink-parent-child-classloading-test/pom.xml
@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>1.5-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-parent-child-classloading-test_${scala.binary.version}</artifactId>
+ <name>flink-parent-child-classloading-test</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>2.4</version>
+
+ <executions>
+ <!-- ClassLoaderTestProgram -->
+ <execution>
+ <id>ClassLoaderTestProgram</id>
+ <phase>package</phase>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ <configuration>
+ <classifier>ClassLoaderTestProgram</classifier>
+
+ <archive>
+ <manifestEntries>
+ <program-class>org.apache.flink.streaming.tests.ClassLoaderTestProgram</program-class>
+ </manifestEntries>
+ </archive>
+
+ <includes>
+ <include>org/apache/flink/streaming/tests/ClassLoaderTestProgram.class</include>
+ <include>org/apache/flink/runtime/taskmanager/TaskManager.class</include>
+ <include>.version.properties</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!--simplify the name of the testing JARs for referring to them in the end-to-end test scripts-->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <version>1.7</version>
+ <executions>
+ <execution>
+ <id>rename</id>
+ <phase>package</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <copy file="${project.basedir}/target/flink-parent-child-classloading-test_${scala.binary.version}-${project.version}-ClassLoaderTestProgram.jar" tofile="${project.basedir}/target/ClassLoaderTestProgram.jar" />
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
+
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
new file mode 100644
index 0000000..3626885
--- /dev/null
+++ b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
@@ -0,0 +1,30 @@
+/*
+ * 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.runtime.taskmanager;
+
+/**
+ * A {@code Taskmanager} in the same package as the proper Flink {@link TaskManager}. We use this
+ * to check whether Flink correctly uses the child-first {@link ClassLoader} when configured to do
+ * so.
+ */
+public class TaskManager {
+ public static String getMessage() {
+ return "Hello, World!";
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
new file mode 100644
index 0000000..1d4ca4c
--- /dev/null
+++ b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
@@ -0,0 +1,102 @@
+/*
+ * 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.streaming.tests;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.taskmanager.TaskManager;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.io.InputStream;
+import java.net.URL;
+import java.util.Enumeration;
+import java.util.Properties;
+
+/**
+ * End-to-end test program for verifying that the {@code classloader.resolve-order} setting
+ * is being honored by Flink. We test this by creating a fake {@code TaskManager} with a single
+ * method that we call in the same package as the original Flink {@code TaskManager} and verify that
+ * we get a {@link NoSuchMethodError} if we're running with {@code parent-first} class loading
+ * and that we get the correct result from the method when we're running with {@code child-first}
+ * class loading.
+ */
+public class ClassLoaderTestProgram {
+
+ public static void main(String[] args) throws Exception {
+
+ final ParameterTool params = ParameterTool.fromArgs(args);
+
+ final String resolveOrder = params.getRequired("resolve-order");
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ env
+ .fromElements("Hello")
+ .map((MapFunction<String, String>) value -> {
+
+ String gitUrl;
+
+ try (InputStream propFile = ClassLoaderTestProgram.class.getClassLoader().getResourceAsStream(".version.properties")) {
+ Properties properties = new Properties();
+ properties.load(propFile);
+ gitUrl = properties.getProperty("git.remote.origin.url");
+ }
+
+ Enumeration<URL> resources = ClassLoaderTestProgram.class.getClassLoader().getResources(
+ ".version.properties");
+
+ StringBuilder sortedProperties = new StringBuilder();
+ while (resources.hasMoreElements()) {
+ URL url = resources.nextElement();
+ try (InputStream in = url.openStream()) {
+ Properties properties = new Properties();
+ properties.load(in);
+ String orderedGitUrl = properties.getProperty("git.remote.origin.url");
+ sortedProperties.append(orderedGitUrl);
+ }
+ }
+
+ if (resolveOrder.equals("parent-first")) {
+ try {
+ @SuppressWarnings("unused")
+ String ignored = TaskManager.getMessage();
+
+ throw new RuntimeException(
+ "TaskManager.getMessage() should not be available with parent-first " +
+ "ClassLoader order.");
+
+ } catch (NoSuchMethodError e) {
+ // expected
+ }
+ return "NoSuchMethodError:" + gitUrl + ":" + sortedProperties;
+ } else if (resolveOrder.equals("child-first")) {
+ String message = TaskManager.getMessage();
+ if (!message.equals("Hello, World!")) {
+ throw new RuntimeException("Wrong message from fake TaskManager.");
+ }
+ return message + ":" + gitUrl + ":" + sortedProperties;
+ } else {
+ throw new RuntimeException("Unknown resolve order: " + resolveOrder);
+ }
+ })
+ .writeAsText(params.getRequired("output"), FileSystem.WriteMode.OVERWRITE);
+
+ env.execute("ClassLoader Test Program");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/resources/.version.properties
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/resources/.version.properties b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/resources/.version.properties
new file mode 100644
index 0000000..dc98aea
--- /dev/null
+++ b/flink-end-to-end-tests/flink-parent-child-classloading-test/src/main/resources/.version.properties
@@ -0,0 +1 @@
+git.remote.origin.url=hello-there-42
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/pom.xml b/flink-end-to-end-tests/pom.xml
index 6e9dee6..695a4f1 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -29,79 +29,13 @@ under the License.
<relativePath>..</relativePath>
</parent>
- <artifactId>flink-end-to-end-tests_${scala.binary.version}</artifactId>
- <name>flink-end-to-end-tests</name>
-
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <version>2.4</version>
+ <packaging>pom</packaging>
- <executions>
- <!-- ClassLoaderTestProgram -->
- <execution>
- <id>ClassLoaderTestProgram</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>ClassLoaderTestProgram</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.streaming.tests.ClassLoaderTestProgram</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>org/apache/flink/streaming/tests/ClassLoaderTestProgram.class</include>
- <include>org/apache/flink/runtime/taskmanager/TaskManager.class</include>
- <include>.version.properties</include>
- </includes>
- </configuration>
- </execution>
- </executions>
- </plugin>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <name>flink-end-to-end-tests</name>
- <!--simplify the name of the testing JARs for referring to them in the end-to-end test scripts-->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <version>1.7</version>
- <executions>
- <execution>
- <id>rename</id>
- <phase>package</phase>
- <goals>
- <goal>run</goal>
- </goals>
- <configuration>
- <target>
- <copy file="${project.basedir}/target/flink-end-to-end-tests_${scala.binary.version}-${project.version}-ClassLoaderTestProgram.jar" tofile="${project.basedir}/target/ClassLoaderTestProgram.jar" />
- </target>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
+ <modules>
+ <module>flink-parent-child-classloading-test</module>
+ </modules>
</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java b/flink-end-to-end-tests/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
deleted file mode 100644
index 3626885..0000000
--- a/flink-end-to-end-tests/src/main/java/org/apache/flink/runtime/taskmanager/TaskManager.java
+++ /dev/null
@@ -1,30 +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.runtime.taskmanager;
-
-/**
- * A {@code Taskmanager} in the same package as the proper Flink {@link TaskManager}. We use this
- * to check whether Flink correctly uses the child-first {@link ClassLoader} when configured to do
- * so.
- */
-public class TaskManager {
- public static String getMessage() {
- return "Hello, World!";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java b/flink-end-to-end-tests/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
deleted file mode 100644
index 1d4ca4c..0000000
--- a/flink-end-to-end-tests/src/main/java/org/apache/flink/streaming/tests/ClassLoaderTestProgram.java
+++ /dev/null
@@ -1,102 +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.streaming.tests;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.runtime.taskmanager.TaskManager;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-
-import java.io.InputStream;
-import java.net.URL;
-import java.util.Enumeration;
-import java.util.Properties;
-
-/**
- * End-to-end test program for verifying that the {@code classloader.resolve-order} setting
- * is being honored by Flink. We test this by creating a fake {@code TaskManager} with a single
- * method that we call in the same package as the original Flink {@code TaskManager} and verify that
- * we get a {@link NoSuchMethodError} if we're running with {@code parent-first} class loading
- * and that we get the correct result from the method when we're running with {@code child-first}
- * class loading.
- */
-public class ClassLoaderTestProgram {
-
- public static void main(String[] args) throws Exception {
-
- final ParameterTool params = ParameterTool.fromArgs(args);
-
- final String resolveOrder = params.getRequired("resolve-order");
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- env
- .fromElements("Hello")
- .map((MapFunction<String, String>) value -> {
-
- String gitUrl;
-
- try (InputStream propFile = ClassLoaderTestProgram.class.getClassLoader().getResourceAsStream(".version.properties")) {
- Properties properties = new Properties();
- properties.load(propFile);
- gitUrl = properties.getProperty("git.remote.origin.url");
- }
-
- Enumeration<URL> resources = ClassLoaderTestProgram.class.getClassLoader().getResources(
- ".version.properties");
-
- StringBuilder sortedProperties = new StringBuilder();
- while (resources.hasMoreElements()) {
- URL url = resources.nextElement();
- try (InputStream in = url.openStream()) {
- Properties properties = new Properties();
- properties.load(in);
- String orderedGitUrl = properties.getProperty("git.remote.origin.url");
- sortedProperties.append(orderedGitUrl);
- }
- }
-
- if (resolveOrder.equals("parent-first")) {
- try {
- @SuppressWarnings("unused")
- String ignored = TaskManager.getMessage();
-
- throw new RuntimeException(
- "TaskManager.getMessage() should not be available with parent-first " +
- "ClassLoader order.");
-
- } catch (NoSuchMethodError e) {
- // expected
- }
- return "NoSuchMethodError:" + gitUrl + ":" + sortedProperties;
- } else if (resolveOrder.equals("child-first")) {
- String message = TaskManager.getMessage();
- if (!message.equals("Hello, World!")) {
- throw new RuntimeException("Wrong message from fake TaskManager.");
- }
- return message + ":" + gitUrl + ":" + sortedProperties;
- } else {
- throw new RuntimeException("Unknown resolve order: " + resolveOrder);
- }
- })
- .writeAsText(params.getRequired("output"), FileSystem.WriteMode.OVERWRITE);
-
- env.execute("ClassLoader Test Program");
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/src/main/resources/.version.properties
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/src/main/resources/.version.properties b/flink-end-to-end-tests/src/main/resources/.version.properties
deleted file mode 100644
index dc98aea..0000000
--- a/flink-end-to-end-tests/src/main/resources/.version.properties
+++ /dev/null
@@ -1 +0,0 @@
-git.remote.origin.url=hello-there-42
http://git-wip-us.apache.org/repos/asf/flink/blob/16b90976/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh b/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
index 95c58f8..34c55f7 100755
--- a/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
+++ b/flink-end-to-end-tests/test-scripts/test_streaming_classloader.sh
@@ -19,7 +19,7 @@
source "$(dirname "$0")"/common.sh
-TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/target/ClassLoaderTestProgram.jar
+TEST_PROGRAM_JAR=$TEST_INFRA_DIR/../../flink-end-to-end-tests/flink-parent-child-classloading-test/target/ClassLoaderTestProgram.jar
echo "Testing parent-first class loading"
@@ -111,4 +111,4 @@ if [[ "$OUTPUT" != "$EXPECTED" ]]; then
echo -e "EXPECTED: $EXPECTED"
echo -e "ACTUAL: $OUTPUT"
PASS=""
-fi
\ No newline at end of file
+fi
[2/8] flink git commit: [FLINK-7343][kafka-tests] Fix test
at-least-once test instability
Posted by ch...@apache.org.
[FLINK-7343][kafka-tests] Fix test at-least-once test instability
Previously we could set numElementsBeforeSnapshot to some value during checkpointing
AFTER executing shutdown, while at the same time FlinkKafkaProducerXXX snapshot for this
value would fail. This lead to incorrectly cacluated expected set of values to be present
in the test kafka topic.
Fix is to remember lastSnapshotedElementBeforeShutdown - last snapshot that we expect
to succeed without failure.
This closes #5729.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0fa76e50
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0fa76e50
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0fa76e50
Branch: refs/heads/release-1.5
Commit: 0fa76e50f06e84da8764089f36227ed26a2c2765
Parents: 29fd6ce
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Tue Mar 20 11:23:35 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:01:31 2018 +0100
----------------------------------------------------------------------
.../connectors/kafka/Kafka09ProducerITCase.java | 11 ----------
.../connectors/kafka/KafkaProducerTestBase.java | 23 ++++++++++----------
2 files changed, 11 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0fa76e50/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ProducerITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ProducerITCase.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ProducerITCase.java
index c619c3e..f145e56 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ProducerITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ProducerITCase.java
@@ -32,15 +32,4 @@ public class Kafka09ProducerITCase extends KafkaProducerTestBase {
public void testExactlyOnceCustomOperator() throws Exception {
// Kafka09 does not support exactly once semantic
}
-
- @Override
- public void testOneToOneAtLeastOnceRegularSink() throws Exception {
- // For some reasons this test is sometimes failing in Kafka09 while the same code works in Kafka010. Disabling
- // this test because everything indicates those failures might be caused by unfixed bugs in Kafka 0.9 branch
- }
-
- @Override
- public void testOneToOneAtLeastOnceCustomOperator() throws Exception {
- // Disable this test since FlinkKafka09Producer doesn't support custom operator mode
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0fa76e50/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 9278b67..5023a7e 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -292,7 +292,7 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
properties,
topic,
partition,
- Collections.unmodifiableSet(new HashSet<>(getIntegersSequence(BrokerRestartingMapper.numElementsBeforeSnapshot))),
+ Collections.unmodifiableSet(new HashSet<>(getIntegersSequence(BrokerRestartingMapper.lastSnapshotedElementBeforeShutdown))),
KAFKA_READ_TIMEOUT);
deleteTestTopic(topic);
@@ -483,7 +483,7 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
private static final long serialVersionUID = 6334389850158707313L;
public static volatile boolean triggeredShutdown;
- public static volatile int numElementsBeforeSnapshot;
+ public static volatile int lastSnapshotedElementBeforeShutdown;
public static volatile Runnable shutdownAction;
private final int failCount;
@@ -493,7 +493,7 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
public static void resetState(Runnable shutdownAction) {
triggeredShutdown = false;
- numElementsBeforeSnapshot = 0;
+ lastSnapshotedElementBeforeShutdown = 0;
BrokerRestartingMapper.shutdownAction = shutdownAction;
}
@@ -509,15 +509,12 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
@Override
public T map(T value) throws Exception {
numElementsTotal++;
+ Thread.sleep(10);
- if (!triggeredShutdown) {
- Thread.sleep(10);
-
- if (failer && numElementsTotal >= failCount) {
- // shut down a Kafka broker
- triggeredShutdown = true;
- shutdownAction.run();
- }
+ if (!triggeredShutdown && failer && numElementsTotal >= failCount) {
+ // shut down a Kafka broker
+ triggeredShutdown = true;
+ shutdownAction.run();
}
return value;
}
@@ -528,7 +525,9 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- numElementsBeforeSnapshot = numElementsTotal;
+ if (!triggeredShutdown) {
+ lastSnapshotedElementBeforeShutdown = numElementsTotal;
+ }
}
@Override
[8/8] flink git commit: [FLINK-8925][tests] Enable flip6 on travis
Posted by ch...@apache.org.
[FLINK-8925][tests] Enable flip6 on travis
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f0e6ae6e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f0e6ae6e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f0e6ae6e
Branch: refs/heads/release-1.5
Commit: f0e6ae6ebfc8eaf1a0a6225b868db0f6fc0f5e60
Parents: c12e21c
Author: zentol <ch...@apache.org>
Authored: Wed Mar 21 15:01:55 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Mar 21 21:04:17 2018 +0100
----------------------------------------------------------------------
.travis.yml | 20 +++++---------------
pom.xml | 7 +++++--
2 files changed, 10 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f0e6ae6e/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index f84b8d8..f86bbef 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -39,27 +39,27 @@ matrix:
- jdk: "oraclejdk8"
env:
- TEST="core"
- - PROFILE="-Dhadoop.version=2.8.3"
+ - PROFILE="-Dhadoop.version=2.8.3 -Dflip6"
- CACHE_NAME=JDK8_H280_CO
- jdk: "oraclejdk8"
env:
- TEST="libraries"
- - PROFILE="-Dhadoop.version=2.8.3"
+ - PROFILE="-Dhadoop.version=2.8.3 -Dflip6"
- CACHE_NAME=JDK8_H280_L
- jdk: "oraclejdk8"
env:
- TEST="connectors"
- - PROFILE="-Dhadoop.version=2.8.3 -Pinclude-kinesis"
+ - PROFILE="-Dhadoop.version=2.8.3 -Dflip6 -Pinclude-kinesis"
- CACHE_NAME=JDK8_H280_CN
- jdk: "oraclejdk8"
env:
- TEST="tests"
- - PROFILE="-Dhadoop.version=2.8.3"
+ - PROFILE="-Dhadoop.version=2.8.3 -Dflip6"
- CACHE_NAME=JDK8_H280_T
- jdk: "oraclejdk8"
env:
- TEST="misc"
- - PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws"
+ - PROFILE="-Dhadoop.version=2.8.3 -Dflip6 -Dinclude_hadoop_aws"
- CACHE_NAME=JDK8_H280_M
- jdk: "openjdk8"
env:
@@ -86,16 +86,6 @@ matrix:
- TEST="misc"
- PROFILE="-Dhadoop.version=2.4.1"
- CACHE_NAME=JDK8_H241_M
- - jdk: "oraclejdk8"
- env:
- - TEST="core"
- - PROFILE="-Dhadoop.version=2.8.0 -Pflip6"
- - CACHE_NAME=JDK8_H280_F6_CO
- - jdk: "oraclejdk8"
- env:
- - TEST="tests"
- - PROFILE="-Dhadoop.version=2.8.0 -Pflip6"
- - CACHE_NAME=JDK8_H280_F6_T
git:
depth: 100
http://git-wip-us.apache.org/repos/asf/flink/blob/f0e6ae6e/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 02ca63b..91e2fd4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -128,7 +128,6 @@ under the License.
<hamcrest.version>1.3</hamcrest.version>
<japicmp.skip>false</japicmp.skip>
<!-- run all groups except flip6 by default -->
- <test.groups></test.groups>
<test.excludedGroups>org.apache.flink.testutils.category.Flip6</test.excludedGroups>
<codebase>old</codebase>
<!--
@@ -595,8 +594,12 @@ under the License.
<profile>
<id>flip6</id>
+ <activation>
+ <property>
+ <name>flip6</name>
+ </property>
+ </activation>
<properties>
- <test.groups>org.apache.flink.testutils.category.Flip6, org.apache.flink.testutils.category.OldAndFlip6</test.groups>
<!-- clear the excluded groups list -->
<test.excludedGroups></test.excludedGroups>
<codebase>flip6</codebase>