You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/10/19 13:23:44 UTC
[GitHub] [flink] twalthr commented on a change in pull request #16142: [FLINK-22748][connector-kafka] Allow dynamic target topic selection in SQL Kafka sinks
twalthr commented on a change in pull request #16142:
URL: https://github.com/apache/flink/pull/16142#discussion_r731854170
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -256,6 +264,254 @@ public void testKafkaTableWithMultipleTopics() throws Exception {
topics.forEach(super::deleteTestTopic);
}
+ @Test
+ public void testKafkaSinkWithMetadataIncludeTopicOption() {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "metadata_topic_" + format;
+ try {
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `physical_1` STRING,\n"
+ + " `physical_2` INT,\n"
+ + " `timestamp` TIMESTAMP(3) METADATA,\n"
+ + " `topic` STRING METADATA,\n"
+ + " `physical_3` BOOLEAN\n"
+ + ") WITH (\n"
+ + " 'connector' = 'kafka',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'earliest-offset',\n"
+ + " %s\n"
+ + ")",
+ topic, bootstraps, groupId, formatOptions());
+
+ tEnv.executeSql(createTable);
+
+ String initialValues =
+ String.format(
+ "INSERT INTO kafka\n"
+ + "VALUES\n"
+ + " ('data 1', 1, TIMESTAMP '2020-03-08 13:12:11.123', '%s', TRUE),\n"
+ + " ('data 2', 2, TIMESTAMP '2020-03-09 13:12:11.123', '%s', FALSE),\n"
+ + " ('data 3', 3, TIMESTAMP '2020-03-10 13:12:11.123', '%s', TRUE)",
+ topic, topic, topic);
+ try {
+ tEnv.executeSql(initialValues).await();
+ fail(
+ "Unable to create the Kafka sink table with table option 'topic' and metadata column 'topic'.");
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
Review comment:
do we really need to create topics for validating an exception? I'm sure we can further simplify this test.
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -256,6 +264,254 @@ public void testKafkaTableWithMultipleTopics() throws Exception {
topics.forEach(super::deleteTestTopic);
}
+ @Test
+ public void testKafkaSinkWithMetadataIncludeTopicOption() {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "metadata_topic_" + format;
+ try {
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `physical_1` STRING,\n"
+ + " `physical_2` INT,\n"
+ + " `timestamp` TIMESTAMP(3) METADATA,\n"
+ + " `topic` STRING METADATA,\n"
+ + " `physical_3` BOOLEAN\n"
+ + ") WITH (\n"
+ + " 'connector' = 'kafka',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'earliest-offset',\n"
+ + " %s\n"
+ + ")",
+ topic, bootstraps, groupId, formatOptions());
+
+ tEnv.executeSql(createTable);
+
+ String initialValues =
+ String.format(
+ "INSERT INTO kafka\n"
+ + "VALUES\n"
+ + " ('data 1', 1, TIMESTAMP '2020-03-08 13:12:11.123', '%s', TRUE),\n"
+ + " ('data 2', 2, TIMESTAMP '2020-03-09 13:12:11.123', '%s', FALSE),\n"
+ + " ('data 3', 3, TIMESTAMP '2020-03-10 13:12:11.123', '%s', TRUE)",
+ topic, topic, topic);
+ try {
+ tEnv.executeSql(initialValues).await();
+ fail(
+ "Unable to create the Kafka sink table with table option 'topic' and metadata column 'topic'.");
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(
+ String.format(
+ "Invalid metadata key '%s' in column 'topic' of table 'default_catalog.default_database.kafka'. "
+ + "The %s class '%s' supports the following metadata keys for writing:\n%s",
+ TOPIC.key,
+ DynamicTableSink.class.getSimpleName(),
+ KafkaDynamicSink.class.getName(),
+ String.join("\n", Arrays.asList(HEADERS.key, TIMESTAMP.key))),
+ e.getMessage());
+ }
+ } finally {
+ // ------------- cleanup -------------------
+
+ deleteTestTopic(topic);
+ }
+ }
+
+ @Test
+ public void testKafkaSinkWithMetadataExcludeTopicOption() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "metadata_topic_" + format;
+ final String test_topic = "test_metadata_topic_" + format;
+ try {
+ createTestTopic(topic, 1, 1);
+ createTestTopic(test_topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
Review comment:
if you use `final` in your code, it should be used where applicable. otherwise a reader thinks this variable is changed again later.
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -256,6 +264,254 @@ public void testKafkaTableWithMultipleTopics() throws Exception {
topics.forEach(super::deleteTestTopic);
}
+ @Test
+ public void testKafkaSinkWithMetadataIncludeTopicOption() {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "metadata_topic_" + format;
+ try {
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `physical_1` STRING,\n"
+ + " `physical_2` INT,\n"
+ + " `timestamp` TIMESTAMP(3) METADATA,\n"
+ + " `topic` STRING METADATA,\n"
+ + " `physical_3` BOOLEAN\n"
+ + ") WITH (\n"
+ + " 'connector' = 'kafka',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'earliest-offset',\n"
+ + " %s\n"
+ + ")",
+ topic, bootstraps, groupId, formatOptions());
+
+ tEnv.executeSql(createTable);
+
+ String initialValues =
+ String.format(
+ "INSERT INTO kafka\n"
+ + "VALUES\n"
+ + " ('data 1', 1, TIMESTAMP '2020-03-08 13:12:11.123', '%s', TRUE),\n"
+ + " ('data 2', 2, TIMESTAMP '2020-03-09 13:12:11.123', '%s', FALSE),\n"
+ + " ('data 3', 3, TIMESTAMP '2020-03-10 13:12:11.123', '%s', TRUE)",
+ topic, topic, topic);
+ try {
+ tEnv.executeSql(initialValues).await();
+ fail(
+ "Unable to create the Kafka sink table with table option 'topic' and metadata column 'topic'.");
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(
+ String.format(
+ "Invalid metadata key '%s' in column 'topic' of table 'default_catalog.default_database.kafka'. "
+ + "The %s class '%s' supports the following metadata keys for writing:\n%s",
+ TOPIC.key,
+ DynamicTableSink.class.getSimpleName(),
+ KafkaDynamicSink.class.getName(),
+ String.join("\n", Arrays.asList(HEADERS.key, TIMESTAMP.key))),
+ e.getMessage());
+ }
+ } finally {
+ // ------------- cleanup -------------------
+
+ deleteTestTopic(topic);
+ }
+ }
+
+ @Test
+ public void testKafkaSinkWithMetadataExcludeTopicOption() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "metadata_topic_" + format;
+ final String test_topic = "test_metadata_topic_" + format;
+ try {
+ createTestTopic(topic, 1, 1);
+ createTestTopic(test_topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createSourceTable =
+ String.format(
+ "CREATE TABLE kafka_source (\n"
+ + " `physical_1` STRING,\n"
+ + " `physical_2` INT,\n"
+ // metadata fields are out of order on purpose
+ // offset is ignored because it might not be deterministic
Review comment:
this comment is not valid anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org