You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by aljoscha <gi...@git.apache.org> on 2017/05/15 20:40:11 UTC
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
GitHub user aljoscha opened a pull request:
https://github.com/apache/flink/pull/3911
[FLINK-6539] Add end-to-end tests
This adds tests that exercise Flink the way a user would:
* start/stop cluster using scripts
* submit job
* Use Flink with a user fat-jar with Kafka 0.10.
This also updates the Kafka Example to use 0.10 and consolidates the two existing examples in one example.
This adds two tests:
* Batch WordCount with verification of results
* Streaming read from Kafka/write to Kafka with verification of results
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/aljoscha/flink test-automation
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3911.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3911
----
commit a79f35bc82b32cf1192da6e3c7e23a4244be896c
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2017-05-09T11:39:27Z
[FLINK-6539] Add automated batch WordCount end-to-end test
commit c3fc1bd037d1b63ea557af28c224e5b0ea770e24
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2017-05-09T12:46:39Z
[FLINK-6539] Combine ReadFromKafka/WriteIntoKafka into one Kafka010Example
This also updates the Kafka version we use in the examples module to
0.10.x.
commit fff2a26a1dd4e268700bed166d77506dafb43dbc
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2017-05-11T10:08:12Z
[FLINK-6539] Add automated Kafka end-to-end test
commit 61e36aee544d806edd05b6fd15993ac47307ddbb
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2017-05-11T10:17:47Z
[FLINK-6539] Run end-to-end tests on travis
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r117691205
--- Diff: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/Kafka010Example.java ---
@@ -17,49 +17,74 @@
package org.apache.flink.streaming.examples.kafka;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
/**
- * Read Strings from Kafka and print them to standard out.
- * Note: On a cluster, DataStream.print() will print to the TaskManager's .out file!
- *
- * Please pass the following arguments to run the example:
- * --topic test --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
+ * An example that shows how to read from and write to Kafka. This will read String messages
+ * from the input topic, prefix them by a configured prefix and output to the output topic.
*
+ * <p>Example usage:
+ * --input-topic test-input --output-topic test-outpu --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
*/
-public class ReadFromKafka {
+public class Kafka010Example {
public static void main(String[] args) throws Exception {
// parse input arguments
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
- if(parameterTool.getNumberOfParameters() < 4) {
- System.out.println("Missing parameters!\nUsage: Kafka --topic <topic> " +
- "--bootstrap.servers <kafka brokers> --zookeeper.connect <zk quorum> --group.id <some id>");
+ if (parameterTool.getNumberOfParameters() < 5) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
+ "--bootstrap.servers <kafka brokers> " +
+ "--zookeeper.connect <zk quorum> --group.id <some id> [--prefix <prefix>]");
return;
}
+ String prefix = parameterTool.get("prefix", "PREFIX:");
+
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
- env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
- env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
+// env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
--- End diff --
It wasn't working because of https://issues.apache.org/jira/browse/FLINK-6515, which is now fixed so will change that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r116659709
--- Diff: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/Kafka010Example.java ---
@@ -17,49 +17,74 @@
package org.apache.flink.streaming.examples.kafka;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
/**
- * Read Strings from Kafka and print them to standard out.
- * Note: On a cluster, DataStream.print() will print to the TaskManager's .out file!
- *
- * Please pass the following arguments to run the example:
- * --topic test --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
+ * An example that shows how to read from and write to Kafka. This will read String messages
+ * from the input topic, prefix them by a configured prefix and output to the output topic.
*
+ * <p>Example usage:
+ * --input-topic test-input --output-topic test-outpu --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
*/
-public class ReadFromKafka {
+public class Kafka010Example {
public static void main(String[] args) throws Exception {
// parse input arguments
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
- if(parameterTool.getNumberOfParameters() < 4) {
- System.out.println("Missing parameters!\nUsage: Kafka --topic <topic> " +
- "--bootstrap.servers <kafka brokers> --zookeeper.connect <zk quorum> --group.id <some id>");
+ if (parameterTool.getNumberOfParameters() < 5) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
+ "--bootstrap.servers <kafka brokers> " +
+ "--zookeeper.connect <zk quorum> --group.id <some id> [--prefix <prefix>]");
return;
}
+ String prefix = parameterTool.get("prefix", "PREFIX:");
+
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
- env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
- env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
+// env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
--- End diff --
By the way, was there any reason to not enable checkpointing in the e2e tests?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
@greghogan No specific reason, to be honest. I took my inspiration for this from Beam, where we also have a `.test-infra` directory: https://github.com/apache/beam. I don't know why we have it there, maybe because normally only Maven sub-modules are visible directories.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by greghogan <gi...@git.apache.org>.
Github user greghogan commented on the issue:
https://github.com/apache/flink/pull/3911
@aljoscha, okay, here the directory is perhaps mistyped `.test-infa`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r117691173
--- Diff: .test-infa/end-to-end-test/test_streaming_kafka010.sh ---
@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+################################################################################
+# 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.
+################################################################################
+
+set -e
+set -o pipefail
+
+# Convert relative path to absolute path
+TEST_ROOT=`pwd`
+TEST_INFRA_DIR="$0"
+TEST_INFRA_DIR=`dirname "$TEST_INFRA_DIR"`
+cd $TEST_INFRA_DIR
+TEST_INFRA_DIR=`pwd`
+cd $TEST_ROOT
+
+. "$TEST_INFRA_DIR"/common.sh
+
+
+start_cluster
+
+# get Kafka 0.10.0
+mkdir -p $TEST_DATA_DIR
+curl "http://mirror.netcologne.de/apache.org/kafka/0.10.2.0/kafka_2.11-0.10.2.0.tgz" > $TEST_DATA_DIR/kafka.tgz
+# Use this when debugging locally, to keep from always downloading
+#cp /Users/aljoscha/Dev/flink/kafka.tgz $TEST_DATA_DIR/kafka.tgz
--- End diff --
Will change that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
Therefore I was waiting for that to get resolved.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r116659749
--- Diff: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/Kafka010Example.java ---
@@ -17,49 +17,74 @@
package org.apache.flink.streaming.examples.kafka;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
/**
- * Read Strings from Kafka and print them to standard out.
- * Note: On a cluster, DataStream.print() will print to the TaskManager's .out file!
- *
- * Please pass the following arguments to run the example:
- * --topic test --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
+ * An example that shows how to read from and write to Kafka. This will read String messages
+ * from the input topic, prefix them by a configured prefix and output to the output topic.
*
+ * <p>Example usage:
+ * --input-topic test-input --output-topic test-outpu --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
*/
-public class ReadFromKafka {
+public class Kafka010Example {
public static void main(String[] args) throws Exception {
// parse input arguments
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
- if(parameterTool.getNumberOfParameters() < 4) {
- System.out.println("Missing parameters!\nUsage: Kafka --topic <topic> " +
- "--bootstrap.servers <kafka brokers> --zookeeper.connect <zk quorum> --group.id <some id>");
+ if (parameterTool.getNumberOfParameters() < 5) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
+ "--bootstrap.servers <kafka brokers> " +
+ "--zookeeper.connect <zk quorum> --group.id <some id> [--prefix <prefix>]");
return;
}
+ String prefix = parameterTool.get("prefix", "PREFIX:");
+
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
- env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
- env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
+// env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(parameterTool);
- DataStream<String> messageStream = env
- .addSource(new FlinkKafkaConsumer08<>(
- parameterTool.getRequired("topic"),
+ DataStream<String> input = env
+ .addSource(new FlinkKafkaConsumer010<>(
+ parameterTool.getRequired("input-topic"),
+ new SimpleStringSchema(),
+ parameterTool.getProperties()))
+ .map(new PrefixingMapper(prefix));
+
+ input.addSink(
+ new FlinkKafkaProducer010<>(
+ parameterTool.getRequired("output-topic"),
new SimpleStringSchema(),
parameterTool.getProperties()));
- // write kafka stream to standard out.
- messageStream.print();
- env.execute("Read from Kafka example");
+ env.execute("Kafka 0.8 Example");
--- End diff --
"0.10 example"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
R: @rmetzger for review, please 🙂
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r117445440
--- Diff: tools/travis_mvn_watchdog.sh ---
@@ -227,5 +227,15 @@ upload_artifacts_s3
# we are going back to
cd ../../
+# run end-to-end tests
+
+echo "Running automated end-to-end tests"
+
+.test-infa/end-to-end-test/test_batch_wordcount.sh build-target cluster
+E2E_WC_EXIT_CODE=$?
+EXIT_CODE=$(($EXIT_CODE+$?))
+.test-infa/end-to-end-test/test_streaming_kafka010.sh build-target cluster
+EXIT_CODE=$(($EXIT_CODE+$?))
--- End diff --
I like the travis integration a lot
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r116659435
--- Diff: .test-infa/end-to-end-test/test_streaming_kafka010.sh ---
@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+################################################################################
+# 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.
+################################################################################
+
+set -e
+set -o pipefail
+
+# Convert relative path to absolute path
+TEST_ROOT=`pwd`
+TEST_INFRA_DIR="$0"
+TEST_INFRA_DIR=`dirname "$TEST_INFRA_DIR"`
+cd $TEST_INFRA_DIR
+TEST_INFRA_DIR=`pwd`
+cd $TEST_ROOT
+
+. "$TEST_INFRA_DIR"/common.sh
+
+
+start_cluster
+
+# get Kafka 0.10.0
+mkdir -p $TEST_DATA_DIR
+curl "http://mirror.netcologne.de/apache.org/kafka/0.10.2.0/kafka_2.11-0.10.2.0.tgz" > $TEST_DATA_DIR/kafka.tgz
+# Use this when debugging locally, to keep from always downloading
+#cp /Users/aljoscha/Dev/flink/kafka.tgz $TEST_DATA_DIR/kafka.tgz
--- End diff --
Should we actually have an option / flag .with binary path for this (instead of going into the script and modifying this)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
Oh oh, that's a typo, yes. I can also change to `test-infra`, since we have several folders that are not maven sub modules already.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
@rmetzger Yes, the test deletes the log files so that additional tests will start from a clean slate. I might be able to modify the Flink config so that each test gets it's own log directory.
Could it be that you already had some log files in your `build-target` Flink installation? This would cause the test to fail if they contain errors/exceptions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/3911
I think this LGTM now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
I think the last kinks are now ironed out. Could you please take another look? (anyone interested)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r116659967
--- Diff: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/WriteIntoKafka.java ---
@@ -1,75 +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.examples.kafka;
-
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer08;
-import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
-
-
-/**
- * Generate a String every 500 ms and write it into a Kafka topic
- *
- * Please pass the following arguments to run the example:
- * --topic test --bootstrap.servers localhost:9092
- *
- */
-public class WriteIntoKafka {
--- End diff --
I understand the intention here, but instead of completely removing this class, shouldn't we just have a new `Kafka010EndToEndExample` instead of replacing the separate write / read examples? Perhaps the end-to-end example should just reuse code from the write / read examples.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3911#discussion_r116659661
--- Diff: flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka/Kafka010Example.java ---
@@ -17,49 +17,74 @@
package org.apache.flink.streaming.examples.kafka;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
/**
- * Read Strings from Kafka and print them to standard out.
- * Note: On a cluster, DataStream.print() will print to the TaskManager's .out file!
- *
- * Please pass the following arguments to run the example:
- * --topic test --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
+ * An example that shows how to read from and write to Kafka. This will read String messages
+ * from the input topic, prefix them by a configured prefix and output to the output topic.
*
+ * <p>Example usage:
+ * --input-topic test-input --output-topic test-outpu --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer
*/
-public class ReadFromKafka {
+public class Kafka010Example {
public static void main(String[] args) throws Exception {
// parse input arguments
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
- if(parameterTool.getNumberOfParameters() < 4) {
- System.out.println("Missing parameters!\nUsage: Kafka --topic <topic> " +
- "--bootstrap.servers <kafka brokers> --zookeeper.connect <zk quorum> --group.id <some id>");
+ if (parameterTool.getNumberOfParameters() < 5) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
+ "--bootstrap.servers <kafka brokers> " +
+ "--zookeeper.connect <zk quorum> --group.id <some id> [--prefix <prefix>]");
return;
}
+ String prefix = parameterTool.get("prefix", "PREFIX:");
+
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
- env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
- env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
+// env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
--- End diff --
Can be removed
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/3911
I tried it locally, but got some weird results:
```
Found error in log files.
Found exception in log files.
Found non-empty file build-target/log/flink-robert-jobmanager-0-robert-da.out
Found non-empty file build-target/log/flink-robert-taskmanager-0-robert-da.out
One or more tests FAILED.
```
It seems that the script deleted the log files. Is that intended?
Also it seemed that WC was running correctly, so I wonder why it failed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by greghogan <gi...@git.apache.org>.
Github user greghogan commented on the issue:
https://github.com/apache/flink/pull/3911
@aljoscha why a hidden directory?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/3911
I think the change is okay to merge.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/3911
Okay
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
Thanks, @tzulitai!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
I think it is, but I was never sure with all the test timeouts.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha closed the pull request at:
https://github.com/apache/flink/pull/3911
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3911: [FLINK-6539] Add end-to-end tests
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/3911
Scratch that, there is still a problem that the tests sometimes fail because of
```
org.apache.kafka.clients.consumer.RetriableCommitFailedException: Commit offsets failed with retriable exception. You should retry committing offsets.
Caused by: org.apache.kafka.common.errors.GroupCoordinatorNotAvailableException: The group coordinator is not available.
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---