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/06/24 12:43:30 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_r657870086
##########
File path: docs/content/docs/connectors/table/kafka.md
##########
@@ -193,10 +193,10 @@ Connector Options
</tr>
<tr>
<td><h5>topic</h5></td>
- <td>required for sink</td>
+ <td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Topic name(s) to read data from when the table is used as source. It also supports topic list for source by separating topic by semicolon like <code>'topic-1;topic-2'</code>. Note, only one of "topic-pattern" and "topic" can be specified for sources. When the table is used as sink, the topic name is the topic to write data to. Note topic list is not supported for sinks.</td>
+ <td>Topic name(s) to read data from when the table is used as source. It also supports topic list for source by separating topic by semicolon like <code>'topic-1;topic-2'</code>. Note, only one of "topic-pattern" and "topic" can be specified for sources. When the table is used as sink, the topic name is the topic to write data. If the "topic" option isn't specified for sink table, "topic" metadata column must be specified. If the "topic" option is specified for sink table, "topic" metadata column must be readable. Note topic list is not supported for sinks.</td>
</tr>
<tr>
Review comment:
I wasn't aware that we support lists already in the topic option. Wouldn't it be useful to support this option to limit the number of topic available in the dynamic metadata column. I think this is a very common use case, because data platform teams usually only allow a set of topics to be writable.
btw `"topic" metadata column must be readable` -> `"topic" metadata column can only be readable`
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
##########
@@ -145,7 +145,7 @@ public KafkaDynamicSink(
// Mutable attributes
this.metadataKeys = Collections.emptyList();
// Kafka-specific attributes
- this.topic = checkNotNull(topic, "Topic must not be null.");
+ this.topic = topic;
Review comment:
mark nullability in parameter using `@Nullable`
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
##########
@@ -216,7 +216,7 @@
protected final Properties producerConfig;
/** The name of the default topic this producer is writing data to. */
- protected final String defaultTopicId;
+ protected String defaultTopicId;
Review comment:
In any case we should make the parameter as `@Nullable` and update the JavaDocs
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
##########
@@ -166,7 +167,16 @@ public void setPartitions(int[] partitions) {
@Override
public String getTargetTopic(RowData element) {
- return topic;
+ if (!StringUtils.isNullOrWhitespaceOnly(topic)) {
Review comment:
this is a very expensive operation that is executed for every record in the hot path, I don't think if we need this whitespace logic at all and can rely on the user to specify it correctly
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
##########
@@ -166,7 +167,16 @@ public void setPartitions(int[] partitions) {
@Override
public String getTargetTopic(RowData element) {
- return topic;
+ if (!StringUtils.isNullOrWhitespaceOnly(topic)) {
+ return topic;
+ }
+ // no explicit topic defined in table options, use the writable metadata column topic
+ final String topicMetadata = readMetadata(element, KafkaDynamicSink.WritableMetadata.TOPIC);
+ if (StringUtils.isNullOrWhitespaceOnly(topicMetadata)) {
+ throw new NullPointerException(
Review comment:
use a different kind of exception, maybe `RuntimeException`
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
##########
@@ -224,7 +224,7 @@ public DynamicTableSink createDynamicTableSink(Context context) {
keyProjection,
valueProjection,
keyPrefix,
- tableOptions.get(TOPIC).get(0),
+ tableOptions.getOptional(TOPIC).isPresent() ? tableOptions.get(TOPIC).get(0) : null,
Review comment:
this could cause a index out of bounds exception if the list is empty?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaSerializationSchema.java
##########
@@ -41,7 +42,7 @@
private final @Nullable FlinkKafkaPartitioner<RowData> partitioner;
- private final String topic;
+ private String topic;
Review comment:
nit: why is it not final anymore? I cannot see another mutation in your changes.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
##########
@@ -355,6 +356,21 @@ public int hashCode() {
// --------------------------------------------------------------------------------------------
enum WritableMetadata {
+ TOPIC(
+ "topic",
+ DataTypes.STRING().nullable(),
Review comment:
shouldn't it be notNull if we throw an exception during runtime otherwise?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaOptions.java
##########
@@ -144,7 +144,7 @@ private KafkaOptions() {}
.noDefaultValue()
.withDescription(
"Topic names from which the table is read. Either 'topic' or 'topic-pattern' must be set for source. "
- + "Option 'topic' is required for sink.");
+ + "Option 'topic' is optional for sink through specifying the 'topic' metadata column.");
Review comment:
This still seems to be different from the docs. Please generate the docs from the options using `ConfigOptionDocGenerator`.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
##########
@@ -216,7 +216,7 @@
protected final Properties producerConfig;
/** The name of the default topic this producer is writing data to. */
- protected final String defaultTopicId;
+ protected String defaultTopicId;
Review comment:
@AHeise you are more familiar with the Kafka producer? Is this a valid change?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org