You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/12/16 11:37:48 UTC
[flink] branch release-1.10 updated: [FLINK-15233][kafka] Improve
Kafka connector properties to make append update-mode as default
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new d58b5cf [FLINK-15233][kafka] Improve Kafka connector properties to make append update-mode as default
d58b5cf is described below
commit d58b5cff36de8c3dbe1151a2eaeb72a41aa77736
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 16 11:04:50 2019 +0800
[FLINK-15233][kafka] Improve Kafka connector properties to make append update-mode as default
This closes #10584
---
docs/dev/table/connect.md | 12 +-----------
docs/dev/table/connect.zh.md | 22 ++++++----------------
.../kafka/KafkaTableSourceSinkFactoryBase.java | 4 ++--
.../flink/table/descriptors/KafkaValidator.java | 5 +++++
.../kafka/KafkaTableSourceSinkFactoryTestBase.java | 9 ++++++++-
5 files changed, 22 insertions(+), 30 deletions(-)
diff --git a/docs/dev/table/connect.md b/docs/dev/table/connect.md
index 0d4dbe6..a3a8dcd 100644
--- a/docs/dev/table/connect.md
+++ b/docs/dev/table/connect.md
@@ -181,9 +181,6 @@ tableEnvironment
.field("message", DataTypes.STRING())
)
- // specify the update-mode for streaming tables
- .inAppendMode()
-
// create a table with given name
.createTemporaryTable("MyUserTable");
{% endhighlight %}
@@ -226,9 +223,7 @@ table_environment \
.field("user", DataTypes.BIGINT())
.field("message", DataTypes.STRING())
) \
- .in_append_mode() \
- .create_temporary_table("MyUserTable")
- # specify the update-mode for streaming tables and
+ .create_temporary_table("MyUserTable")
# register as source, sink, or both and under a name
{% endhighlight %}
</div>
@@ -238,7 +233,6 @@ table_environment \
tables:
- name: MyUserTable # name the new table
type: source # declare if the table should be "source", "sink", or "both"
- update-mode: append # specify the update-mode for streaming tables
# declare the external system to connect to
connector:
@@ -297,7 +291,6 @@ CREATE TABLE MyUserTable (
'connector.startup-mode' = 'earliest-offset',
'connector.properties.zookeeper.connect' = 'localhost:2181',
'connector.properties.bootstrap.servers' = 'localhost:9092',
- 'update-mode' = 'append',
-- declare a format for this system
'format.type' = 'avro',
'format.avro-schema' = '{
@@ -791,9 +784,6 @@ CREATE TABLE MyUserTable (
'connector.topic' = 'topic_name', -- required: topic name from which the table is read
- 'update-mode' = 'append', -- required: update mode when used as table sink,
- -- only support append mode now.
-
'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
diff --git a/docs/dev/table/connect.zh.md b/docs/dev/table/connect.zh.md
index fa3b301..67bac63 100644
--- a/docs/dev/table/connect.zh.md
+++ b/docs/dev/table/connect.zh.md
@@ -181,9 +181,6 @@ tableEnvironment
.field("message", DataTypes.STRING())
)
- // specify the update-mode for streaming tables
- .inAppendMode()
-
// create a table with given name
.createTemporaryTable("MyUserTable");
{% endhighlight %}
@@ -226,9 +223,7 @@ table_environment \
.field("user", DataTypes.BIGINT())
.field("message", DataTypes.STRING())
) \
- .in_append_mode() \
.create_temporary_table("MyUserTable")
- # specify the update-mode for streaming tables and
# register as source, sink, or both and under a name
{% endhighlight %}
</div>
@@ -238,7 +233,6 @@ table_environment \
tables:
- name: MyUserTable # name the new table
type: source # declare if the table should be "source", "sink", or "both"
- update-mode: append # specify the update-mode for streaming tables
# declare the external system to connect to
connector:
@@ -297,7 +291,6 @@ CREATE TABLE MyUserTable (
'connector.startup-mode' = 'earliest-offset',
'connector.properties.zookeeper.connect' = 'localhost:2181',
'connector.properties.bootstrap.servers' = 'localhost:9092',
- 'update-mode' = 'append',
-- declare a format for this system
'format.type' = 'avro',
'format.avro-schema' = '{
@@ -762,7 +755,7 @@ connector:
# "0.8", "0.9", "0.10", "0.11", and "universal"
topic: ... # required: topic name from which the table is read
- properties:
+ properties:
zookeeper.connect: localhost:2181 # required: specify the ZooKeeper connection string
bootstrap.servers: localhost:9092 # required: specify the Kafka server connection string
group.id: testGroup # optional: required in Kafka consumer, specify consumer group
@@ -770,7 +763,7 @@ connector:
startup-mode: ... # optional: valid modes are "earliest-offset", "latest-offset",
# "group-offsets", or "specific-offsets"
specific-offsets: partition:0,offset:42;partition:1,offset:300 # optional: used in case of startup mode with specific offsets
-
+
sink-partitioner: ... # optional: output partitioning from Flink's partitions into Kafka's partitions
# valid are "fixed" (each Flink partition ends up in at most one Kafka partition),
# "round-robin" (a Flink partition is distributed to Kafka partitions round-robin)
@@ -791,18 +784,15 @@ CREATE TABLE MyUserTable (
'connector.topic' = 'topic_name', -- required: topic name from which the table is read
- 'update-mode' = 'append', -- required: update mode when used as table sink,
- -- only support append mode now.
-
- 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specifies the ZooKeeper connection string
- 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specifies the Kafka server connection string
- 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specifies consumer group
+ 'connector.properties.zookeeper.connect' = 'localhost:2181', -- required: specify the ZooKeeper connection string
+ 'connector.properties.bootstrap.servers' = 'localhost:9092', -- required: specify the Kafka server connection string
+ 'connector.properties.group.id' = 'testGroup', --optional: required in Kafka consumer, specify consumer group
'connector.startup-mode' = 'earliest-offset', -- optional: valid modes are "earliest-offset",
-- "latest-offset", "group-offsets",
-- or "specific-offsets"
-- optional: used in case of startup mode with specific offsets
- 'connector.specific-offsets' = 'partition:0,offset:42,partition:1,offset:300',
+ 'connector.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300',
'connector.sink-partitioner' = '...', -- optional: output partitioning from Flink's partitions
-- into Kafka's partitions valid are "fixed"
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
index f1881ee..a9fb480 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
@@ -81,7 +81,6 @@ import static org.apache.flink.table.descriptors.Schema.SCHEMA_NAME;
import static org.apache.flink.table.descriptors.Schema.SCHEMA_PROCTIME;
import static org.apache.flink.table.descriptors.Schema.SCHEMA_TYPE;
import static org.apache.flink.table.descriptors.StreamTableDescriptorValidator.UPDATE_MODE;
-import static org.apache.flink.table.descriptors.StreamTableDescriptorValidator.UPDATE_MODE_VALUE_APPEND;
/**
* Factory for creating configured instances of {@link KafkaTableSourceBase}.
@@ -93,7 +92,6 @@ public abstract class KafkaTableSourceSinkFactoryBase implements
@Override
public Map<String, String> requiredContext() {
Map<String, String> context = new HashMap<>();
- context.put(UPDATE_MODE, UPDATE_MODE_VALUE_APPEND); // append mode
context.put(CONNECTOR_TYPE, CONNECTOR_TYPE_VALUE_KAFKA); // kafka
context.put(CONNECTOR_VERSION, kafkaVersion()); // version
context.put(CONNECTOR_PROPERTY_VERSION, "1"); // backwards compatibility
@@ -103,6 +101,8 @@ public abstract class KafkaTableSourceSinkFactoryBase implements
@Override
public List<String> supportedProperties() {
List<String> properties = new ArrayList<>();
+ // update mode
+ properties.add(UPDATE_MODE);
// kafka
properties.add(CONNECTOR_TOPIC);
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
index 98a68ea..24ecfef 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java
@@ -22,11 +22,14 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.table.api.ValidationException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
import static org.apache.flink.table.descriptors.DescriptorProperties.noValidation;
+import static org.apache.flink.table.descriptors.StreamTableDescriptorValidator.UPDATE_MODE;
+import static org.apache.flink.table.descriptors.StreamTableDescriptorValidator.UPDATE_MODE_VALUE_APPEND;
/**
* The validator for {@link Kafka}.
@@ -64,6 +67,8 @@ public class KafkaValidator extends ConnectorDescriptorValidator {
@Override
public void validate(DescriptorProperties properties) {
super.validate(properties);
+ properties.validateEnumValues(UPDATE_MODE, true, Collections.singletonList(UPDATE_MODE_VALUE_APPEND));
+
properties.validateValue(CONNECTOR_TYPE, CONNECTOR_TYPE_VALUE_KAFKA, false);
properties.validateString(CONNECTOR_TOPIC, false, 1, Integer.MAX_VALUE);
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
index fbc38a7..f51e0dc 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java
@@ -213,6 +213,10 @@ public abstract class KafkaTableSourceSinkFactoryTestBase extends TestLogger {
legacyPropertiesMap.remove("connector.properties.bootstrap.servers");
legacyPropertiesMap.remove("connector.properties.group.id");
+ // keep compatible with a specified update-mode
+ legacyPropertiesMap.put("update-mode", "append");
+
+ // legacy properties for specific-offsets and properties
legacyPropertiesMap.put("connector.specific-offsets.0.partition", "0");
legacyPropertiesMap.put("connector.specific-offsets.0.offset", "100");
legacyPropertiesMap.put("connector.specific-offsets.1.partition", "1");
@@ -252,7 +256,6 @@ public abstract class KafkaTableSourceSinkFactoryTestBase extends TestLogger {
.field(EVENT_TIME, DataTypes.TIMESTAMP(3)).rowtime(
new Rowtime().timestampsFromField(TIME).watermarksPeriodicAscending())
.field(PROC_TIME, DataTypes.TIMESTAMP(3)).proctime())
- .inAppendMode()
.toProperties();
}
@@ -316,6 +319,10 @@ public abstract class KafkaTableSourceSinkFactoryTestBase extends TestLogger {
legacyPropertiesMap.remove("connector.properties.bootstrap.servers");
legacyPropertiesMap.remove("connector.properties.group.id");
+ // keep compatible with a specified update-mode
+ legacyPropertiesMap.put("update-mode", "append");
+
+ // legacy properties for specific-offsets and properties
legacyPropertiesMap.put("connector.specific-offsets.0.partition", "0");
legacyPropertiesMap.put("connector.specific-offsets.0.offset", "100");
legacyPropertiesMap.put("connector.specific-offsets.1.partition", "1");