You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/06/28 16:35:37 UTC
[1/2] flink git commit: [FLINK-8983] End-to-end test: Confluent
schema registry
Repository: flink
Updated Branches:
refs/heads/master da37daa8b -> 9b3660456
[FLINK-8983] End-to-end test: Confluent schema registry
This closes #6083.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a36b5699
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a36b5699
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a36b5699
Branch: refs/heads/master
Commit: a36b56999240d1ead0793be7acb4ad13cd0559f2
Parents: da37daa
Author: Yadan.JS <y_...@yahoo.com>
Authored: Tue Jun 12 10:03:58 2018 -0400
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Jun 28 13:43:50 2018 +0200
----------------------------------------------------------------------
.../flink-confluent-schema-registry/pom.xml | 155 +++++++++++++++++++
.../src/main/avro/user.avsc | 27 ++++
.../test/TestAvroConsumerConfluent.java | 86 ++++++++++
flink-end-to-end-tests/pom.xml | 1 +
flink-end-to-end-tests/run-nightly-tests.sh | 2 +
.../test-scripts/kafka-common.sh | 29 ++++
.../test-confluent-schema-registry.sh | 106 +++++++++++++
7 files changed, 406 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
new file mode 100644
index 0000000..576fca6
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
@@ -0,0 +1,155 @@
+<!--
+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.6-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-confluent-schema-registry</artifactId>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <confluent.version>4.1.0</confluent.version>
+ </properties>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>http://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <!-- Apache Flink dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <!-- This dependency is required to actually execute jobs. It is currently pulled in by
+ flink-streaming-java, but we explicitly depend on it to safeguard against future changes. -->
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients_${scala.binary.version}</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>
+ <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.10 -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-avro</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-avro-confluent-registry</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.0.0</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <finalName>TestAvroConsumerConfluent</finalName>
+ <artifactSet>
+ <excludes>
+ <exclude>com.google.code.findbugs:jsr305</exclude>
+ </excludes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.flink.schema.registry.test.TestAvroConsumerConfluent</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro-maven-plugin</artifactId>
+ <version>${avro.version}</version>
+ <executions>
+ <execution>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>schema</goal>
+ </goals>
+ <configuration>
+ <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
+ <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
+ <fieldVisibility>PRIVATE</fieldVisibility>
+ <includes>
+ <include>**/*.avsc</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>dependency-convergence</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/avro/user.avsc
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/avro/user.avsc b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/avro/user.avsc
new file mode 100644
index 0000000..aca9b83
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/avro/user.avsc
@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+
+ {"namespace": "example.avro",
+ "type": "record",
+ "name": "User",
+ "fields": [
+ {"name": "name", "type": "string", "default": ""},
+ {"name": "favoriteNumber", "type": "string", "default": ""},
+ {"name": "favoriteColor", "type": "string", "default": ""},
+ {"name": "eventType","type": {"name": "EventType","type": "enum", "symbols": ["meeting"] }}
+ ]
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
new file mode 100644
index 0000000..9149832
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
@@ -0,0 +1,86 @@
+/*
+ * 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.schema.registry.test;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
+
+import example.avro.User;
+
+import java.util.Properties;
+
+/**
+ * A simple example that shows how to read from and write to Kafka with Confluent Schema Registry.
+ * This will read AVRO messages from the input topic, parse them into a POJO type via checking the Schema by calling Schema registry.
+ * Then this example publish the POJO type to kafka by converting the POJO to AVRO and verifying the schema.
+ * --input-topic test-input --output-topic test-output --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --schema-registry-url http://localhost:8081 --group.id myconsumer
+ */
+public class TestAvroConsumerConfluent {
+
+ public static void main(String[] args) throws Exception {
+ Properties config = new Properties();
+ // parse input arguments
+ final ParameterTool parameterTool = ParameterTool.fromArgs(args);
+
+ if (parameterTool.getNumberOfParameters() < 6) {
+ System.out.println("Missing parameters!\n" +
+ "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
+ "--bootstrap.servers <kafka brokers> " +
+ "--zookeeper.connect <zk quorum> " +
+ "--schema-registry-url <confluent schema registry> --group.id <some id>");
+ return;
+ }
+ config.setProperty("bootstrap.servers", parameterTool.getRequired("bootstrap.servers"));
+ config.setProperty("group.id", parameterTool.getRequired("group.id"));
+ config.setProperty("zookeeper.connect", parameterTool.getRequired("zookeeper.connect"));
+ String schemaRegistryUrl = parameterTool.getRequired("schema-registry-url");
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().disableSysoutLogging();
+
+ DataStreamSource<User> input = env
+ .addSource(
+ new FlinkKafkaConsumer010(
+ parameterTool.getRequired("input-topic"),
+ ConfluentRegistryAvroDeserializationSchema.forSpecific(User.class, schemaRegistryUrl),
+ config).setStartFromEarliest());
+
+ SingleOutputStreamOperator<String> mapToString = input
+ .map(new MapFunction<User, String>() {
+ @Override
+ public String map(User value) throws Exception {
+ return value.toString();
+ }
+ });
+
+ FlinkKafkaProducer010<String> stringFlinkKafkaProducer010 = new FlinkKafkaProducer010(
+ parameterTool.getRequired("output-topic"),
+ new SimpleStringSchema(),
+ config);
+
+ mapToString.addSink(stringFlinkKafkaProducer010);
+ env.execute("Kafka 0.10 Confluent Schema Registry AVRO Example");
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/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 c169050..091dd0b 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -49,6 +49,7 @@ under the License.
<module>flink-elasticsearch2-test</module>
<module>flink-elasticsearch5-test</module>
<module>flink-quickstart-test</module>
+ <module>flink-confluent-schema-registry</module>
</modules>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/run-nightly-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 302e27f..5873349 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -98,5 +98,7 @@ run_test "Elasticsearch (v5.1.2) sink end-to-end test" "$END_TO_END_DIR/test-scr
run_test "Quickstarts Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh java"
run_test "Quickstarts Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh scala"
+run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test-confluent-schema-registry.sh"
+
printf "\n[PASS] All tests passed\n"
exit 0
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/test-scripts/kafka-common.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/kafka-common.sh b/flink-end-to-end-tests/test-scripts/kafka-common.sh
index 49ff4fe..50fba12 100644
--- a/flink-end-to-end-tests/test-scripts/kafka-common.sh
+++ b/flink-end-to-end-tests/test-scripts/kafka-common.sh
@@ -25,6 +25,7 @@ if [[ -z $TEST_DATA_DIR ]]; then
fi
KAFKA_DIR=$TEST_DATA_DIR/kafka_2.11-0.10.2.0
+CONFLUENT_DIR=$TEST_DATA_DIR/confluent-3.2.0
function setup_kafka_dist {
# download Kafka
@@ -40,6 +41,16 @@ function setup_kafka_dist {
sed -i -e "s+^\(log\.dirs\s*=\s*\).*$+\1$TEST_DATA_DIR/kafka+" $KAFKA_DIR/config/server.properties
}
+function setup_confluent_dist {
+ # download confluent
+ mkdir -p $TEST_DATA_DIR
+ CONFLUENT_URL="http://packages.confluent.io/archive/3.2/confluent-oss-3.2.0-2.11.tar.gz"
+ echo "Downloading confluent from $CONFLUENT_URL"
+ curl "$CONFLUENT_URL" > $TEST_DATA_DIR/confluent.tgz
+
+ tar xzf $TEST_DATA_DIR/confluent.tgz -C $TEST_DATA_DIR/
+}
+
function start_kafka_cluster {
if [[ -z $KAFKA_DIR ]]; then
echo "Must run 'setup_kafka_dist' before attempting to start Kafka cluster"
@@ -76,6 +87,16 @@ function read_messages_from_kafka {
--consumer-property group.id=$3 2> /dev/null
}
+function read_messages_from_kafka_avro {
+ $CONFLUENT_DIR/bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --from-beginning \
+ --max-messages $1 \
+ --topic $2 2> /dev/null
+}
+
+function send_messages_to_kafka_avro {
+echo -e $1 | $CONFLUENT_DIR/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic $2 --property value.schema=$3
+}
+
function modify_num_partitions {
$KAFKA_DIR/bin/kafka-topics.sh --alter --topic $1 --partitions $2 --zookeeper localhost:2181
}
@@ -97,3 +118,11 @@ function get_partition_end_offset {
| tr -s " " \
| cut -d " " -f 4
}
+
+function start_confluent_schema_registry {
+ $CONFLUENT_DIR/bin/schema-registry-start -daemon $CONFLUENT_DIR/etc/schema-registry/schema-registry.properties
+}
+
+function stop_confluent_schema_registry {
+ $CONFLUENT_DIR/bin/schema-registry-stop
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/a36b5699/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh b/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
new file mode 100755
index 0000000..5cebb72
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
@@ -0,0 +1,106 @@
+#!/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.
+################################################################################
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/kafka-common.sh
+
+function verify_output {
+ local expected=$(printf $1)
+ local result=$(echo $2 | sed 's/ //g')
+
+ if [[ "$result" != "$expected" ]]; then
+ echo "Output from Flink program does not match expected output."
+ echo -e "EXPECTED FOR KEY: --$expected--"
+ echo -e "ACTUAL: --$result--"
+ PASS=""
+ exit 1
+ fi
+}
+
+function test_cleanup {
+ # don't call ourselves again for another signal interruption
+ trap "exit -1" INT
+ # don't call ourselves again for normal exit
+ trap "" EXIT
+
+ stop_kafka_cluster
+ stop_confluent_schema_registry
+
+ # revert our modifications to the Flink distribution
+ mv -f $FLINK_DIR/conf/flink-conf.yaml.bak $FLINK_DIR/conf/flink-conf.yaml
+
+ # make sure to run regular cleanup as well
+ cleanup
+}
+
+trap test_cleanup INT
+trap test_cleanup EXIT
+
+setup_kafka_dist
+setup_confluent_dist
+
+cd flink-end-to-end-tests/flink-confluent-schema-registry
+mvn clean package -nsu
+
+start_kafka_cluster
+start_confluent_schema_registry
+sleep 5
+
+# modify configuration to use port 8082 for Flink
+cp $FLINK_DIR/conf/flink-conf.yaml $FLINK_DIR/conf/flink-conf.yaml.bak
+sed -i -e "s/web.port: 8081/web.port: 8082/" $FLINK_DIR/conf/flink-conf.yaml
+
+TEST_PROGRAM_JAR=target/TestAvroConsumerConfluent.jar
+
+INPUT_MESSAGE_1='{"name":"Alyssa","favoriteNumber":"250","favoriteColor":"green","eventType":"meeting"}'
+INPUT_MESSAGE_2='{"name":"Charlie","favoriteNumber":"10","favoriteColor":"blue","eventType":"meeting"}'
+INPUT_MESSAGE_3='{"name":"Ben","favoriteNumber":"7","favoriteColor":"red","eventType":"meeting"}'
+USER_SCHEMA='{"namespace":"example.avro","type":"record","name":"User","fields":[{"name":"name","type":"string","default":""},{"name":"favoriteNumber","type":"string","default":""},{"name":"favoriteColor","type":"string","default":""},{"name":"eventType","type":{"name":"EventType","type":"enum","symbols":["meeting"]}}]}'
+
+curl -X POST \
+ http://localhost:8081/subjects/users-value/versions \
+ -H 'cache-control: no-cache' \
+ -H 'content-type: application/vnd.schemaregistry.v1+json' \
+ -d '{"schema": "{\"namespace\": \"example.avro\",\"type\": \"record\",\"name\": \"User\",\"fields\": [{\"name\": \"name\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteNumber\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteColor\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"eventType\",\"type\": {\"name\": \"EventType\",\"type\": \"enum\", \"symbols\": [\"meeting\"] }}]}"}'
+
+echo "Sending messages to Kafka topic [test-avro-input] ..."
+
+send_messages_to_kafka_avro $INPUT_MESSAGE_1 test-avro-input $USER_SCHEMA
+send_messages_to_kafka_avro $INPUT_MESSAGE_2 test-avro-input $USER_SCHEMA
+send_messages_to_kafka_avro $INPUT_MESSAGE_3 test-avro-input $USER_SCHEMA
+
+start_cluster
+
+# Read Avro message from [test-avro-input], check the schema and send message to [test-avro-out]
+$FLINK_DIR/bin/flink run -d $TEST_PROGRAM_JAR \
+ --input-topic test-avro-input --output-topic test-avro-out \
+ --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer --auto.offset.reset earliest \
+ --schema-registry-url http://localhost:8081
+
+#echo "Reading messages from Kafka topic [test-avro-out] ..."
+
+KEY_1_MSGS=$(read_messages_from_kafka 3 test-avro-out Alyssa_consumer | grep Alyssa)
+KEY_2_MSGS=$(read_messages_from_kafka 3 test-avro-out Charlie_consumer | grep Charlie)
+KEY_3_MSGS=$(read_messages_from_kafka 3 test-avro-out Ben_consumer | grep Ben)
+
+## Verifying AVRO output with actual message
+verify_output $INPUT_MESSAGE_1 "$KEY_1_MSGS"
+verify_output $INPUT_MESSAGE_2 "$KEY_2_MSGS"
+verify_output $INPUT_MESSAGE_3 "$KEY_3_MSGS"
+
[2/2] flink git commit: [FLINK-8983] Integrate
test_confluent_schema_registry.sh into run-nightly-tests.sh
Posted by tr...@apache.org.
[FLINK-8983] Integrate test_confluent_schema_registry.sh into run-nightly-tests.sh
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9b366045
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9b366045
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9b366045
Branch: refs/heads/master
Commit: 9b366045697f80534b8eb2e8b559f02d1452f0cf
Parents: a36b569
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Jun 28 18:34:24 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Jun 28 18:34:24 2018 +0200
----------------------------------------------------------------------
flink-end-to-end-tests/.gitignore | 1 +
.../flink-confluent-schema-registry/.gitignore | 1 +
.../flink-confluent-schema-registry/pom.xml | 7 ++
.../test/TestAvroConsumerConfluent.java | 14 +--
flink-end-to-end-tests/run-nightly-tests.sh | 2 +-
.../test-scripts/kafka-common.sh | 28 +++--
.../test-confluent-schema-registry.sh | 106 -------------------
.../test_confluent_schema_registry.sh | 93 ++++++++++++++++
8 files changed, 128 insertions(+), 124 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/.gitignore
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/.gitignore b/flink-end-to-end-tests/.gitignore
new file mode 100644
index 0000000..3fc3a72
--- /dev/null
+++ b/flink-end-to-end-tests/.gitignore
@@ -0,0 +1 @@
+test-scripts/temp-test-directory*
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/flink-confluent-schema-registry/.gitignore
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/.gitignore b/flink-end-to-end-tests/flink-confluent-schema-registry/.gitignore
new file mode 100644
index 0000000..b83a752
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/.gitignore
@@ -0,0 +1 @@
+src/main/java/example/avro
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
index 576fca6..2f63c22 100644
--- a/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
@@ -150,6 +150,13 @@ under the License.
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <configuration>
+ <excludes>**/example/avro/*</excludes>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
index 9149832..fbbe3c3 100644
--- a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java
@@ -28,6 +28,7 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import example.avro.User;
+import org.apache.avro.specific.SpecificRecordBase;
import java.util.Properties;
@@ -40,7 +41,6 @@ import java.util.Properties;
public class TestAvroConsumerConfluent {
public static void main(String[] args) throws Exception {
- Properties config = new Properties();
// parse input arguments
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
@@ -52,6 +52,7 @@ public class TestAvroConsumerConfluent {
"--schema-registry-url <confluent schema registry> --group.id <some id>");
return;
}
+ Properties config = new Properties();
config.setProperty("bootstrap.servers", parameterTool.getRequired("bootstrap.servers"));
config.setProperty("group.id", parameterTool.getRequired("group.id"));
config.setProperty("zookeeper.connect", parameterTool.getRequired("zookeeper.connect"));
@@ -62,20 +63,15 @@ public class TestAvroConsumerConfluent {
DataStreamSource<User> input = env
.addSource(
- new FlinkKafkaConsumer010(
+ new FlinkKafkaConsumer010<>(
parameterTool.getRequired("input-topic"),
ConfluentRegistryAvroDeserializationSchema.forSpecific(User.class, schemaRegistryUrl),
config).setStartFromEarliest());
SingleOutputStreamOperator<String> mapToString = input
- .map(new MapFunction<User, String>() {
- @Override
- public String map(User value) throws Exception {
- return value.toString();
- }
- });
+ .map((MapFunction<User, String>) SpecificRecordBase::toString);
- FlinkKafkaProducer010<String> stringFlinkKafkaProducer010 = new FlinkKafkaProducer010(
+ FlinkKafkaProducer010<String> stringFlinkKafkaProducer010 = new FlinkKafkaProducer010<>(
parameterTool.getRequired("output-topic"),
new SimpleStringSchema(),
config);
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/run-nightly-tests.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 5873349..11eeafe 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -98,7 +98,7 @@ run_test "Elasticsearch (v5.1.2) sink end-to-end test" "$END_TO_END_DIR/test-scr
run_test "Quickstarts Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh java"
run_test "Quickstarts Scala nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_quickstarts.sh scala"
-run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test-confluent-schema-registry.sh"
+run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_confluent_schema_registry.sh"
printf "\n[PASS] All tests passed\n"
exit 0
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/test-scripts/kafka-common.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/kafka-common.sh b/flink-end-to-end-tests/test-scripts/kafka-common.sh
index 50fba12..f0a37b8 100644
--- a/flink-end-to-end-tests/test-scripts/kafka-common.sh
+++ b/flink-end-to-end-tests/test-scripts/kafka-common.sh
@@ -26,6 +26,8 @@ fi
KAFKA_DIR=$TEST_DATA_DIR/kafka_2.11-0.10.2.0
CONFLUENT_DIR=$TEST_DATA_DIR/confluent-3.2.0
+SCHEMA_REGISTRY_PORT=8082
+SCHEMA_REGISTRY_URL=http://localhost:${SCHEMA_REGISTRY_PORT}
function setup_kafka_dist {
# download Kafka
@@ -49,6 +51,9 @@ function setup_confluent_dist {
curl "$CONFLUENT_URL" > $TEST_DATA_DIR/confluent.tgz
tar xzf $TEST_DATA_DIR/confluent.tgz -C $TEST_DATA_DIR/
+
+ # fix confluent config
+ sed -i -e "s#listeners=http://0.0.0.0:8081#listeners=http://0.0.0.0:${SCHEMA_REGISTRY_PORT}#" $CONFLUENT_DIR/etc/schema-registry/schema-registry.properties
}
function start_kafka_cluster {
@@ -87,14 +92,8 @@ function read_messages_from_kafka {
--consumer-property group.id=$3 2> /dev/null
}
-function read_messages_from_kafka_avro {
- $CONFLUENT_DIR/bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --from-beginning \
- --max-messages $1 \
- --topic $2 2> /dev/null
-}
-
function send_messages_to_kafka_avro {
-echo -e $1 | $CONFLUENT_DIR/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic $2 --property value.schema=$3
+echo -e $1 | $CONFLUENT_DIR/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic $2 --property value.schema=$3 --property schema.registry.url=${SCHEMA_REGISTRY_URL}
}
function modify_num_partitions {
@@ -120,7 +119,20 @@ function get_partition_end_offset {
}
function start_confluent_schema_registry {
- $CONFLUENT_DIR/bin/schema-registry-start -daemon $CONFLUENT_DIR/etc/schema-registry/schema-registry.properties
+ $CONFLUENT_DIR/bin/schema-registry-start -daemon $CONFLUENT_DIR/etc/schema-registry/schema-registry.properties
+
+ # wait until the schema registry REST endpoint is up
+ for i in {1..30}; do
+ QUERY_RESULT=$(curl "${SCHEMA_REGISTRY_URL}/subjects" 2> /dev/null || true)
+
+ if [[ ${QUERY_RESULT} =~ \[.*\] ]]; then
+ echo "Schema registry is up."
+ break
+ fi
+
+ echo "Waiting for schema registry..."
+ sleep 1
+ done
}
function stop_confluent_schema_registry {
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh b/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
deleted file mode 100755
index 5cebb72..0000000
--- a/flink-end-to-end-tests/test-scripts/test-confluent-schema-registry.sh
+++ /dev/null
@@ -1,106 +0,0 @@
-#!/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.
-################################################################################
-
-source "$(dirname "$0")"/common.sh
-source "$(dirname "$0")"/kafka-common.sh
-
-function verify_output {
- local expected=$(printf $1)
- local result=$(echo $2 | sed 's/ //g')
-
- if [[ "$result" != "$expected" ]]; then
- echo "Output from Flink program does not match expected output."
- echo -e "EXPECTED FOR KEY: --$expected--"
- echo -e "ACTUAL: --$result--"
- PASS=""
- exit 1
- fi
-}
-
-function test_cleanup {
- # don't call ourselves again for another signal interruption
- trap "exit -1" INT
- # don't call ourselves again for normal exit
- trap "" EXIT
-
- stop_kafka_cluster
- stop_confluent_schema_registry
-
- # revert our modifications to the Flink distribution
- mv -f $FLINK_DIR/conf/flink-conf.yaml.bak $FLINK_DIR/conf/flink-conf.yaml
-
- # make sure to run regular cleanup as well
- cleanup
-}
-
-trap test_cleanup INT
-trap test_cleanup EXIT
-
-setup_kafka_dist
-setup_confluent_dist
-
-cd flink-end-to-end-tests/flink-confluent-schema-registry
-mvn clean package -nsu
-
-start_kafka_cluster
-start_confluent_schema_registry
-sleep 5
-
-# modify configuration to use port 8082 for Flink
-cp $FLINK_DIR/conf/flink-conf.yaml $FLINK_DIR/conf/flink-conf.yaml.bak
-sed -i -e "s/web.port: 8081/web.port: 8082/" $FLINK_DIR/conf/flink-conf.yaml
-
-TEST_PROGRAM_JAR=target/TestAvroConsumerConfluent.jar
-
-INPUT_MESSAGE_1='{"name":"Alyssa","favoriteNumber":"250","favoriteColor":"green","eventType":"meeting"}'
-INPUT_MESSAGE_2='{"name":"Charlie","favoriteNumber":"10","favoriteColor":"blue","eventType":"meeting"}'
-INPUT_MESSAGE_3='{"name":"Ben","favoriteNumber":"7","favoriteColor":"red","eventType":"meeting"}'
-USER_SCHEMA='{"namespace":"example.avro","type":"record","name":"User","fields":[{"name":"name","type":"string","default":""},{"name":"favoriteNumber","type":"string","default":""},{"name":"favoriteColor","type":"string","default":""},{"name":"eventType","type":{"name":"EventType","type":"enum","symbols":["meeting"]}}]}'
-
-curl -X POST \
- http://localhost:8081/subjects/users-value/versions \
- -H 'cache-control: no-cache' \
- -H 'content-type: application/vnd.schemaregistry.v1+json' \
- -d '{"schema": "{\"namespace\": \"example.avro\",\"type\": \"record\",\"name\": \"User\",\"fields\": [{\"name\": \"name\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteNumber\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteColor\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"eventType\",\"type\": {\"name\": \"EventType\",\"type\": \"enum\", \"symbols\": [\"meeting\"] }}]}"}'
-
-echo "Sending messages to Kafka topic [test-avro-input] ..."
-
-send_messages_to_kafka_avro $INPUT_MESSAGE_1 test-avro-input $USER_SCHEMA
-send_messages_to_kafka_avro $INPUT_MESSAGE_2 test-avro-input $USER_SCHEMA
-send_messages_to_kafka_avro $INPUT_MESSAGE_3 test-avro-input $USER_SCHEMA
-
-start_cluster
-
-# Read Avro message from [test-avro-input], check the schema and send message to [test-avro-out]
-$FLINK_DIR/bin/flink run -d $TEST_PROGRAM_JAR \
- --input-topic test-avro-input --output-topic test-avro-out \
- --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer --auto.offset.reset earliest \
- --schema-registry-url http://localhost:8081
-
-#echo "Reading messages from Kafka topic [test-avro-out] ..."
-
-KEY_1_MSGS=$(read_messages_from_kafka 3 test-avro-out Alyssa_consumer | grep Alyssa)
-KEY_2_MSGS=$(read_messages_from_kafka 3 test-avro-out Charlie_consumer | grep Charlie)
-KEY_3_MSGS=$(read_messages_from_kafka 3 test-avro-out Ben_consumer | grep Ben)
-
-## Verifying AVRO output with actual message
-verify_output $INPUT_MESSAGE_1 "$KEY_1_MSGS"
-verify_output $INPUT_MESSAGE_2 "$KEY_2_MSGS"
-verify_output $INPUT_MESSAGE_3 "$KEY_3_MSGS"
-
http://git-wip-us.apache.org/repos/asf/flink/blob/9b366045/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
----------------------------------------------------------------------
diff --git a/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh b/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
new file mode 100755
index 0000000..be0b014
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
@@ -0,0 +1,93 @@
+#!/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.
+################################################################################
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/kafka-common.sh
+
+function verify_output {
+ local expected=$(printf $1)
+ local result=$(echo $2 | sed 's/ //g')
+
+ if [[ "$result" != "$expected" ]]; then
+ echo "Output from Flink program does not match expected output."
+ echo -e "EXPECTED FOR KEY: --$expected--"
+ echo -e "ACTUAL: --$result--"
+ exit 1
+ fi
+}
+
+function test_cleanup {
+ # don't call ourselves again for another signal interruption
+ trap "exit -1" INT
+ # don't call ourselves again for normal exit
+ trap "" EXIT
+
+ stop_kafka_cluster
+ stop_confluent_schema_registry
+}
+
+trap test_cleanup INT
+trap test_cleanup EXIT
+
+setup_kafka_dist
+setup_confluent_dist
+
+start_kafka_cluster
+start_confluent_schema_registry
+
+TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-confluent-schema-registry/target/TestAvroConsumerConfluent.jar
+
+INPUT_MESSAGE_1='{"name":"Alyssa","favoriteNumber":"250","favoriteColor":"green","eventType":"meeting"}'
+INPUT_MESSAGE_2='{"name":"Charlie","favoriteNumber":"10","favoriteColor":"blue","eventType":"meeting"}'
+INPUT_MESSAGE_3='{"name":"Ben","favoriteNumber":"7","favoriteColor":"red","eventType":"meeting"}'
+USER_SCHEMA='{"namespace":"example.avro","type":"record","name":"User","fields":[{"name":"name","type":"string","default":""},{"name":"favoriteNumber","type":"string","default":""},{"name":"favoriteColor","type":"string","default":""},{"name":"eventType","type":{"name":"EventType","type":"enum","symbols":["meeting"]}}]}'
+
+curl -X POST \
+ ${SCHEMA_REGISTRY_URL}/subjects/users-value/versions \
+ -H 'cache-control: no-cache' \
+ -H 'content-type: application/vnd.schemaregistry.v1+json' \
+ -d '{"schema": "{\"namespace\": \"example.avro\",\"type\": \"record\",\"name\": \"User\",\"fields\": [{\"name\": \"name\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteNumber\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"favoriteColor\", \"type\": \"string\", \"default\": \"\"},{\"name\": \"eventType\",\"type\": {\"name\": \"EventType\",\"type\": \"enum\", \"symbols\": [\"meeting\"] }}]}"}'
+
+echo "Sending messages to Kafka topic [test-avro-input] ..."
+
+send_messages_to_kafka_avro $INPUT_MESSAGE_1 test-avro-input $USER_SCHEMA
+send_messages_to_kafka_avro $INPUT_MESSAGE_2 test-avro-input $USER_SCHEMA
+send_messages_to_kafka_avro $INPUT_MESSAGE_3 test-avro-input $USER_SCHEMA
+
+start_cluster
+
+create_kafka_topic 1 1 test-avro-out
+
+# Read Avro message from [test-avro-input], check the schema and send message to [test-avro-out]
+$FLINK_DIR/bin/flink run -d $TEST_PROGRAM_JAR \
+ --input-topic test-avro-input --output-topic test-avro-out \
+ --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer --auto.offset.reset earliest \
+ --schema-registry-url ${SCHEMA_REGISTRY_URL}
+
+#echo "Reading messages from Kafka topic [test-avro-out] ..."
+
+KEY_1_MSGS=$(read_messages_from_kafka 3 test-avro-out Alyssa_consumer | grep Alyssa)
+KEY_2_MSGS=$(read_messages_from_kafka 3 test-avro-out Charlie_consumer | grep Charlie)
+KEY_3_MSGS=$(read_messages_from_kafka 3 test-avro-out Ben_consumer | grep Ben)
+
+## Verifying AVRO output with actual message
+verify_output $INPUT_MESSAGE_1 "$KEY_1_MSGS"
+verify_output $INPUT_MESSAGE_2 "$KEY_2_MSGS"
+verify_output $INPUT_MESSAGE_3 "$KEY_3_MSGS"
+