You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2018/11/15 10:58:28 UTC
[flink] branch release-1.7 updated (b6f7551 -> 2f5435f)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git.
from b6f7551 [FLINK-10767] Updated savepoint compatibility matrix with 1.7
new 62f9fbb [FLINK-10624] Extend SQL client end-to-end to test new KafkaTableSink
new 2f5435f [FLINK-10861]][e2e] Do not fail stop_kafka_cluster if a process terminates before we kill it
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink-sql-client-test/pom.xml | 17 +-
flink-end-to-end-tests/run-nightly-tests.sh | 2 +
.../test-scripts/kafka-common.sh | 4 +-
.../test-scripts/kafka_sql_common.sh | 101 +++++++++
.../test-scripts/test_sql_client.sh | 229 +++------------------
.../config/ssh_config => test_sql_client_kafka.sh} | 9 +-
.../ssh_config => test_sql_client_kafka010.sh} | 9 +-
.../test-scripts/test_sql_client_kafka_common.sh | 220 ++++++++++++++++++++
8 files changed, 379 insertions(+), 212 deletions(-)
create mode 100644 flink-end-to-end-tests/test-scripts/kafka_sql_common.sh
copy flink-end-to-end-tests/test-scripts/{docker-hadoop-secure-cluster/config/ssh_config => test_sql_client_kafka.sh} (88%)
mode change 100644 => 100755
copy flink-end-to-end-tests/test-scripts/{docker-hadoop-secure-cluster/config/ssh_config => test_sql_client_kafka010.sh} (87%)
mode change 100644 => 100755
create mode 100755 flink-end-to-end-tests/test-scripts/test_sql_client_kafka_common.sh
[flink] 01/02: [FLINK-10624] Extend SQL client end-to-end to test
new KafkaTableSink
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 62f9fbbdaa00650dec21da7f4b32c3e383ac0a8d
Author: yanghua <ya...@gmail.com>
AuthorDate: Tue Nov 13 22:00:01 2018 +0800
[FLINK-10624] Extend SQL client end-to-end to test new KafkaTableSink
This closes #6927.
---
.../flink-sql-client-test/pom.xml | 17 +-
flink-end-to-end-tests/run-nightly-tests.sh | 2 +
.../test-scripts/kafka_sql_common.sh | 101 +++++++++
.../test-scripts/test_sql_client.sh | 229 +++------------------
.../test-scripts/test_sql_client_kafka.sh | 22 ++
.../test-scripts/test_sql_client_kafka010.sh | 22 ++
.../test-scripts/test_sql_client_kafka_common.sh | 220 ++++++++++++++++++++
7 files changed, 413 insertions(+), 200 deletions(-)
diff --git a/flink-end-to-end-tests/flink-sql-client-test/pom.xml b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
index de1db9e..7710d01 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/pom.xml
+++ b/flink-end-to-end-tests/flink-sql-client-test/pom.xml
@@ -92,6 +92,14 @@ under the License.
<dependency>
<!-- Used by maven-dependency-plugin -->
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <classifier>sql-jar</classifier>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <!-- Used by maven-dependency-plugin -->
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<classifier>sql-jar</classifier>
@@ -106,7 +114,7 @@ under the License.
as we neither access nor package the kafka dependencies -->
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
- <version>0.11.0.2</version>
+ <version>2.0.0</version>
</dependency>
</dependencies>
</dependencyManagement>
@@ -176,6 +184,13 @@ under the License.
<classifier>sql-jar</classifier>
<type>jar</type>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <classifier>sql-jar</classifier>
+ <type>jar</type>
+ </artifactItem>
<!-- This SQL JAR is not used for now to avoid dependency conflicts; see FLINK-10107.
<artifactItem>
<groupId>org.apache.flink</groupId>
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 7e15870..6d45608 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -142,6 +142,8 @@ run_test "State TTL Heap backend end-to-end test" "$END_TO_END_DIR/test-scripts/
run_test "State TTL RocksDb backend end-to-end test" "$END_TO_END_DIR/test-scripts/test_stream_state_ttl.sh rocks"
run_test "SQL Client end-to-end test" "$END_TO_END_DIR/test-scripts/test_sql_client.sh"
+run_test "SQL Client end-to-end test for Kafka 0.10" "$END_TO_END_DIR/test-scripts/test_sql_client_kafka010.sh"
+run_test "SQL Client end-to-end test for modern Kafka" "$END_TO_END_DIR/test-scripts/test_sql_client_kafka.sh"
run_test "Heavy deployment end-to-end test" "$END_TO_END_DIR/test-scripts/test_heavy_deployment.sh"
diff --git a/flink-end-to-end-tests/test-scripts/kafka_sql_common.sh b/flink-end-to-end-tests/test-scripts/kafka_sql_common.sh
new file mode 100644
index 0000000..3e95b40
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/kafka_sql_common.sh
@@ -0,0 +1,101 @@
+#!/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.
+################################################################################
+
+KAFKA_CONNECTOR_VERSION="$1"
+KAFKA_VERSION="$2"
+CONFLUENT_VERSION="$3"
+CONFLUENT_MAJOR_VERSION="$4"
+
+source "$(dirname "$0")"/kafka-common.sh $2 $3 $4
+
+function create_kafka_json_source {
+ topicName="$1"
+ create_kafka_topic 1 1 $topicName
+
+ # put JSON data into Kafka
+ echo "Sending messages to Kafka..."
+
+ send_messages_to_kafka '{"timestamp": "2018-03-12 08:00:00", "user": "Alice", "event": { "type": "WARNING", "message": "This is a warning."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 08:10:00", "user": "Alice", "event": { "type": "WARNING", "message": "This is a warning."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 09:00:00", "user": "Bob", "event": { "type": "WARNING", "message": "This is another warning."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 09:10:00", "user": "Alice", "event": { "type": "INFO", "message": "This is a info."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 09:20:00", "user": "Steve", "event": { "type": "INFO", "message": "This is another info."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 09:30:00", "user": "Steve", "event": { "type": "INFO", "message": "This is another info."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 09:30:00", "user": null, "event": { "type": "WARNING", "message": "This is a bad message because the user is missing."}}' $topicName
+ send_messages_to_kafka '{"timestamp": "2018-03-12 10:40:00", "user": "Bob", "event": { "type": "ERROR", "message": "This is an error."}}' $topicName
+}
+
+function get_kafka_json_source_schema {
+ topicName="$1"
+ tableName="$2"
+ cat << EOF
+ - name: $tableName
+ type: source-table
+ update-mode: append
+ schema:
+ - name: rowtime
+ type: TIMESTAMP
+ rowtime:
+ timestamps:
+ type: from-field
+ from: timestamp
+ watermarks:
+ type: periodic-bounded
+ delay: 2000
+ - name: user
+ type: VARCHAR
+ - name: event
+ type: ROW<type VARCHAR, message VARCHAR>
+ connector:
+ type: kafka
+ version: "$KAFKA_CONNECTOR_VERSION"
+ topic: $topicName
+ startup-mode: earliest-offset
+ properties:
+ - key: zookeeper.connect
+ value: localhost:2181
+ - key: bootstrap.servers
+ value: localhost:9092
+ format:
+ type: json
+ json-schema: >
+ {
+ "type": "object",
+ "properties": {
+ "timestamp": {
+ "type": "string"
+ },
+ "user": {
+ "type": ["string", "null"]
+ },
+ "event": {
+ "type": "object",
+ "properties": {
+ "type": {
+ "type": "string"
+ },
+ "message": {
+ "type": "string"
+ }
+ }
+ }
+ }
+ }
+EOF
+}
diff --git a/flink-end-to-end-tests/test-scripts/test_sql_client.sh b/flink-end-to-end-tests/test-scripts/test_sql_client.sh
index b9d0eca..74192b4 100755
--- a/flink-end-to-end-tests/test-scripts/test_sql_client.sh
+++ b/flink-end-to-end-tests/test-scripts/test_sql_client.sh
@@ -19,8 +19,13 @@
set -Eeuo pipefail
+KAFKA_CONNECTOR_VERSION="2.0"
+KAFKA_VERSION="2.0.0"
+CONFLUENT_VERSION="5.0.0"
+CONFLUENT_MAJOR_VERSION="5.0"
+
source "$(dirname "$0")"/common.sh
-source "$(dirname "$0")"/kafka-common.sh 0.10.2.0 3.2.0 3.2
+source "$(dirname "$0")"/kafka_sql_common.sh $KAFKA_CONNECTOR_VERSION $KAFKA_VERSION $CONFLUENT_VERSION $CONFLUENT_MAJOR_VERSION
source "$(dirname "$0")"/elasticsearch-common.sh
SQL_TOOLBOX_JAR=$END_TO_END_DIR/flink-sql-client-test/target/SqlToolbox.jar
@@ -87,25 +92,9 @@ setup_kafka_dist
start_kafka_cluster
-create_kafka_topic 1 1 test-json
+create_kafka_json_source test-json
create_kafka_topic 1 1 test-avro
-# put JSON data into Kafka
-echo "Sending messages to Kafka..."
-
-send_messages_to_kafka '{"timestamp": "2018-03-12 08:00:00", "user": "Alice", "event": { "type": "WARNING", "message": "This is a warning."}}' test-json
-# duplicate
-send_messages_to_kafka '{"timestamp": "2018-03-12 08:10:00", "user": "Alice", "event": { "type": "WARNING", "message": "This is a warning."}}' test-json
-send_messages_to_kafka '{"timestamp": "2018-03-12 09:00:00", "user": "Bob", "event": { "type": "WARNING", "message": "This is another warning."}}' test-json
-send_messages_to_kafka '{"timestamp": "2018-03-12 09:10:00", "user": "Alice", "event": { "type": "INFO", "message": "This is a info."}}' test-json
-send_messages_to_kafka '{"timestamp": "2018-03-12 09:20:00", "user": "Steve", "event": { "type": "INFO", "message": "This is another info."}}' test-json
-# duplicate
-send_messages_to_kafka '{"timestamp": "2018-03-12 09:30:00", "user": "Steve", "event": { "type": "INFO", "message": "This is another info."}}' test-json
-# filtered in results
-send_messages_to_kafka '{"timestamp": "2018-03-12 09:30:00", "user": null, "event": { "type": "WARNING", "message": "This is a bad message because the user is missing."}}' test-json
-# pending in results
-send_messages_to_kafka '{"timestamp": "2018-03-12 10:40:00", "user": "Bob", "event": { "type": "ERROR", "message": "This is an error."}}' test-json
-
# prepare Elasticsearch
echo "Preparing Elasticsearch..."
@@ -116,141 +105,35 @@ setup_elasticsearch $DOWNLOAD_URL
wait_elasticsearch_working
################################################################################
-# Run a SQL statement
+# Prepare Flink
################################################################################
-echo "Testing SQL statement..."
-
-# prepare Flink
echo "Preparing Flink..."
start_cluster
start_taskmanagers 2
+################################################################################
+# Run SQL statements
+################################################################################
+
+echo "Testing SQL statements..."
+
+JSON_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "json" )
+KAFKA_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "kafka_" )
+ELASTICSEARCH_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "elasticsearch6" )
+
# create session environment file
RESULT=$TEST_DATA_DIR/result
SQL_CONF=$TEST_DATA_DIR/sql-client-session.conf
-cat > $SQL_CONF << EOF
+cat >> $SQL_CONF << EOF
tables:
- - name: JsonSourceTable
- type: source-table
- update-mode: append
- schema:
- - name: rowtime
- type: TIMESTAMP
- rowtime:
- timestamps:
- type: from-field
- from: timestamp
- watermarks:
- type: periodic-bounded
- delay: 2000
- - name: user
- type: VARCHAR
- - name: event
- type: ROW<type VARCHAR, message VARCHAR>
- connector:
- type: kafka
- version: "0.10"
- topic: test-json
- startup-mode: earliest-offset
- properties:
- - key: zookeeper.connect
- value: localhost:2181
- - key: bootstrap.servers
- value: localhost:9092
- format:
- type: json
- json-schema: >
- {
- "type": "object",
- "properties": {
- "timestamp": {
- "type": "string"
- },
- "user": {
- "type": ["string", "null"]
- },
- "event": {
- "type": "object",
- "properties": {
- "type": {
- "type": "string"
- },
- "message": {
- "type": "string"
- }
- }
- }
- }
- }
- - name: AvroBothTable
- type: source-sink-table
- update-mode: append
- schema:
- - name: event_timestamp
- type: VARCHAR
- - name: user
- type: VARCHAR
- - name: message
- type: VARCHAR
- - name: duplicate_count
- type: BIGINT
- connector:
- type: kafka
- version: "0.10"
- topic: test-avro
- startup-mode: earliest-offset
- properties:
- - key: zookeeper.connect
- value: localhost:2181
- - key: bootstrap.servers
- value: localhost:9092
- format:
- type: avro
- avro-schema: >
- {
- "namespace": "org.apache.flink.table.tests",
- "type": "record",
- "name": "NormalizedEvent",
- "fields": [
- {"name": "event_timestamp", "type": "string"},
- {"name": "user", "type": ["string", "null"]},
- {"name": "message", "type": "string"},
- {"name": "duplicate_count", "type": "long"}
- ]
- }
- - name: CsvSinkTable
- type: sink-table
- update-mode: append
- schema:
- - name: event_timestamp
- type: VARCHAR
- - name: user
- type: VARCHAR
- - name: message
- type: VARCHAR
- - name: duplicate_count
- type: BIGINT
- - name: constant
- type: VARCHAR
- connector:
- type: filesystem
- path: $RESULT
- format:
- type: csv
- fields:
- - name: event_timestamp
- type: VARCHAR
- - name: user
- type: VARCHAR
- - name: message
- type: VARCHAR
- - name: duplicate_count
- type: BIGINT
- - name: constant
- type: VARCHAR
+EOF
+
+get_kafka_json_source_schema test-json JsonSourceTable >> $SQL_CONF
+
+cat >> $SQL_CONF << EOF
- name: ElasticsearchUpsertSinkTable
type: sink-table
update-mode: upsert
@@ -308,62 +191,6 @@ EOF
# submit SQL statements
-echo "Executing SQL: Kafka JSON -> Kafka Avro"
-
-SQL_STATEMENT_1=$(cat << EOF
-INSERT INTO AvroBothTable
- SELECT
- CAST(TUMBLE_START(rowtime, INTERVAL '1' HOUR) AS VARCHAR) AS event_timestamp,
- user,
- RegReplace(event.message, ' is ', ' was ') AS message,
- COUNT(*) AS duplicate_count
- FROM JsonSourceTable
- WHERE user IS NOT NULL
- GROUP BY
- user,
- event.message,
- TUMBLE(rowtime, INTERVAL '1' HOUR)
-EOF
-)
-
-echo "$SQL_STATEMENT_1"
-
-$FLINK_DIR/bin/sql-client.sh embedded \
- --library $SQL_JARS_DIR \
- --jar $SQL_TOOLBOX_JAR \
- --environment $SQL_CONF \
- --update "$SQL_STATEMENT_1"
-
-echo "Executing SQL: Kafka Avro -> Filesystem CSV"
-
-SQL_STATEMENT_2=$(cat << EOF
-INSERT INTO CsvSinkTable
- SELECT AvroBothTable.*, RegReplace('Test constant folding.', 'Test', 'Success') AS constant
- FROM AvroBothTable
-EOF
-)
-
-echo "$SQL_STATEMENT_2"
-
-$FLINK_DIR/bin/sql-client.sh embedded \
- --library $SQL_JARS_DIR \
- --jar $SQL_TOOLBOX_JAR \
- --environment $SQL_CONF \
- --update "$SQL_STATEMENT_2"
-
-echo "Waiting for CSV results..."
-for i in {1..10}; do
- if [ -e $RESULT ]; then
- CSV_LINE_COUNT=`cat $RESULT | wc -l`
- if [ $((CSV_LINE_COUNT)) -eq 4 ]; then
- break
- fi
- fi
- sleep 5
-done
-
-check_result_hash "SQL Client Kafka" $RESULT "0a1bf8bf716069b7269f575f87a802c0"
-
echo "Executing SQL: Values -> Elasticsearch (upsert)"
SQL_STATEMENT_3=$(cat << EOF
@@ -403,7 +230,9 @@ EOF
)
JOB_ID=$($FLINK_DIR/bin/sql-client.sh embedded \
- --library $SQL_JARS_DIR \
+ --jar $KAFKA_SQL_JAR \
+ --jar $JSON_SQL_JAR \
+ --jar $ELASTICSEARCH_SQL_JAR \
--jar $SQL_TOOLBOX_JAR \
--environment $SQL_CONF \
--update "$SQL_STATEMENT_4" | grep "Job ID:" | sed 's/.* //g')
@@ -434,7 +263,9 @@ EOF
)
JOB_ID=$($FLINK_DIR/bin/sql-client.sh embedded \
- --library $SQL_JARS_DIR \
+ --jar $KAFKA_SQL_JAR \
+ --jar $JSON_SQL_JAR \
+ --jar $ELASTICSEARCH_SQL_JAR \
--jar $SQL_TOOLBOX_JAR \
--environment $SQL_CONF \
--update "$SQL_STATEMENT_5" | grep "Job ID:" | sed 's/.* //g')
diff --git a/flink-end-to-end-tests/test-scripts/test_sql_client_kafka.sh b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka.sh
new file mode 100755
index 0000000..a05dc05
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka.sh
@@ -0,0 +1,22 @@
+#!/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 -Eeuo pipefail
+
+source "$(dirname "$0")"/test_sql_client_kafka_common.sh 2.0 2.0.0 5.0.0 5.0 "kafka"
diff --git a/flink-end-to-end-tests/test-scripts/test_sql_client_kafka010.sh b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka010.sh
new file mode 100755
index 0000000..c710abc
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka010.sh
@@ -0,0 +1,22 @@
+#!/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 -Eeuo pipefail
+
+source "$(dirname "$0")"/test_sql_client_kafka_common.sh 0.10 0.10.2.0 3.2.0 3.2 "kafka-0.10"
diff --git a/flink-end-to-end-tests/test-scripts/test_sql_client_kafka_common.sh b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka_common.sh
new file mode 100755
index 0000000..149c86f
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/test_sql_client_kafka_common.sh
@@ -0,0 +1,220 @@
+#!/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.
+################################################################################
+
+KAFKA_CONNECTOR_VERSION="$1"
+KAFKA_VERSION="$2"
+CONFLUENT_VERSION="$3"
+CONFLUENT_MAJOR_VERSION="$4"
+KAFKA_SQL_JAR="$5"
+
+source "$(dirname "$0")"/common.sh
+source "$(dirname "$0")"/kafka_sql_common.sh $KAFKA_CONNECTOR_VERSION $KAFKA_VERSION $CONFLUENT_VERSION $CONFLUENT_MAJOR_VERSION
+
+################################################################################
+# Prepare connectors
+################################################################################
+
+function sql_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
+}
+trap sql_cleanup INT
+trap sql_cleanup EXIT
+
+# prepare Kafka
+echo "Preparing Kafka $KAFKA_VERSION..."
+
+setup_kafka_dist
+
+start_kafka_cluster
+
+create_kafka_json_source test-json
+create_kafka_topic 1 1 test-avro
+
+################################################################################
+# Prepare Flink
+################################################################################
+
+echo "Preparing Flink..."
+
+start_cluster
+start_taskmanagers 2
+
+################################################################################
+# Run SQL statements
+################################################################################
+
+echo "Testing SQL statements..."
+
+SQL_TOOLBOX_JAR=$END_TO_END_DIR/flink-sql-client-test/target/SqlToolbox.jar
+SQL_JARS_DIR=$END_TO_END_DIR/flink-sql-client-test/target/sql-jars
+
+AVRO_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "avro" )
+JSON_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "json" )
+KAFKA_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "${KAFKA_SQL_JAR}_" )
+
+# create session environment file
+RESULT=$TEST_DATA_DIR/result
+SQL_CONF=$TEST_DATA_DIR/sql-client-session.conf
+
+cat >> $SQL_CONF << EOF
+tables:
+EOF
+
+get_kafka_json_source_schema test-json JsonSourceTable >> $SQL_CONF
+
+cat >> $SQL_CONF << EOF
+ - name: AvroBothTable
+ type: source-sink-table
+ update-mode: append
+ schema:
+ - name: event_timestamp
+ type: VARCHAR
+ - name: user
+ type: VARCHAR
+ - name: message
+ type: VARCHAR
+ - name: duplicate_count
+ type: BIGINT
+ connector:
+ type: kafka
+ version: "$KAFKA_CONNECTOR_VERSION"
+ topic: test-avro
+ startup-mode: earliest-offset
+ properties:
+ - key: zookeeper.connect
+ value: localhost:2181
+ - key: bootstrap.servers
+ value: localhost:9092
+ format:
+ type: avro
+ avro-schema: >
+ {
+ "namespace": "org.apache.flink.table.tests",
+ "type": "record",
+ "name": "NormalizedEvent",
+ "fields": [
+ {"name": "event_timestamp", "type": "string"},
+ {"name": "user", "type": ["string", "null"]},
+ {"name": "message", "type": "string"},
+ {"name": "duplicate_count", "type": "long"}
+ ]
+ }
+ - name: CsvSinkTable
+ type: sink-table
+ update-mode: append
+ schema:
+ - name: event_timestamp
+ type: VARCHAR
+ - name: user
+ type: VARCHAR
+ - name: message
+ type: VARCHAR
+ - name: duplicate_count
+ type: BIGINT
+ - name: constant
+ type: VARCHAR
+ connector:
+ type: filesystem
+ path: $RESULT
+ format:
+ type: csv
+ fields:
+ - name: event_timestamp
+ type: VARCHAR
+ - name: user
+ type: VARCHAR
+ - name: message
+ type: VARCHAR
+ - name: duplicate_count
+ type: BIGINT
+ - name: constant
+ type: VARCHAR
+
+functions:
+ - name: RegReplace
+ from: class
+ class: org.apache.flink.table.toolbox.StringRegexReplaceFunction
+EOF
+
+# submit SQL statements
+
+echo "Executing SQL: Kafka $KAFKA_VERSION JSON -> Kafka $KAFKA_VERSION Avro"
+
+SQL_STATEMENT_1=$(cat << EOF
+INSERT INTO AvroBothTable
+ SELECT
+ CAST(TUMBLE_START(rowtime, INTERVAL '1' HOUR) AS VARCHAR) AS event_timestamp,
+ user,
+ RegReplace(event.message, ' is ', ' was ') AS message,
+ COUNT(*) AS duplicate_count
+ FROM JsonSourceTable
+ WHERE user IS NOT NULL
+ GROUP BY
+ user,
+ event.message,
+ TUMBLE(rowtime, INTERVAL '1' HOUR)
+EOF
+)
+
+echo "$SQL_STATEMENT_1"
+
+$FLINK_DIR/bin/sql-client.sh embedded \
+ --jar $KAFKA_SQL_JAR \
+ --jar $JSON_SQL_JAR \
+ --jar $AVRO_SQL_JAR \
+ --jar $SQL_TOOLBOX_JAR \
+ --environment $SQL_CONF \
+ --update "$SQL_STATEMENT_1"
+
+echo "Executing SQL: Kafka $KAFKA_VERSION Avro -> Filesystem CSV"
+
+SQL_STATEMENT_2=$(cat << EOF
+INSERT INTO CsvSinkTable
+ SELECT AvroBothTable.*, RegReplace('Test constant folding.', 'Test', 'Success') AS constant
+ FROM AvroBothTable
+EOF
+)
+
+echo "$SQL_STATEMENT_2"
+
+$FLINK_DIR/bin/sql-client.sh embedded \
+ --jar $KAFKA_SQL_JAR \
+ --jar $JSON_SQL_JAR \
+ --jar $AVRO_SQL_JAR \
+ --jar $SQL_TOOLBOX_JAR \
+ --environment $SQL_CONF \
+ --update "$SQL_STATEMENT_2"
+
+echo "Waiting for CSV results..."
+for i in {1..10}; do
+ if [ -e $RESULT ]; then
+ CSV_LINE_COUNT=`cat $RESULT | wc -l`
+ if [ $((CSV_LINE_COUNT)) -eq 4 ]; then
+ break
+ fi
+ fi
+ sleep 5
+done
+
+check_result_hash "SQL Client Kafka $KAFKA_VERSION" $RESULT "0a1bf8bf716069b7269f575f87a802c0"
[flink] 02/02: [FLINK-10861]][e2e] Do not fail stop_kafka_cluster
if a process terminates before we kill it
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2f5435fdef73e91aa115a95e19d221eb045bd27b
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Nov 13 11:24:35 2018 +0100
[FLINK-10861]][e2e] Do not fail stop_kafka_cluster if a process terminates before we kill it
---
flink-end-to-end-tests/test-scripts/kafka-common.sh | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 5f8f65e..7c31534 100644
--- a/flink-end-to-end-tests/test-scripts/kafka-common.sh
+++ b/flink-end-to-end-tests/test-scripts/kafka-common.sh
@@ -82,14 +82,14 @@ function stop_kafka_cluster {
PIDS=$(jps -vl | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}'|| echo "")
if [ ! -z "$PIDS" ]; then
- kill -s TERM $PIDS
+ kill -s TERM $PIDS || true
fi
# Terminate QuorumPeerMain process if it still exists
PIDS=$(jps -vl | grep java | grep -i QuorumPeerMain | grep -v grep | awk '{print $1}'|| echo "")
if [ ! -z "$PIDS" ]; then
- kill -s TERM $PIDS
+ kill -s TERM $PIDS || true
fi
}