klam-shop commented on code in PR #109:
URL:
https://github.com/apache/flink-connector-kafka/pull/109#discussion_r1690254530
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java:
##########
@@ -144,14 +147,34 @@ public void open(
valueSerialization.open(context);
}
+ private String getTargetTopic(RowData element) {
+ final String topic = readMetadata(element,
KafkaDynamicSink.WritableMetadata.TOPIC);
+ if (topic == null && topics == null) {
+ throw new IllegalArgumentException(
+ "The topic of the sink record is not valid. Expected a
single topic but no topic is set.");
+ } else if (topic == null && topics.size() == 1) {
+ return topics.get(0);
Review Comment:
Good idea, I thought about it and thought it might make the semantics a bit
complicated, because if we do this, then for `topics.size == 1` then I think
the `topic` metadata column should not be writable, but I think it's worth
saving the processing. Updated
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java:
##########
@@ -144,14 +147,34 @@ public void open(
valueSerialization.open(context);
}
+ private String getTargetTopic(RowData element) {
+ final String topic = readMetadata(element,
KafkaDynamicSink.WritableMetadata.TOPIC);
+ if (topic == null && topics == null) {
+ throw new IllegalArgumentException(
+ "The topic of the sink record is not valid. Expected a
single topic but no topic is set.");
+ } else if (topic == null && topics.size() == 1) {
+ return topics.get(0);
Review Comment:
Good idea, I thought about it before and thought it might make the semantics
a bit complicated, because if we do this, then for `topics.size == 1` then I
think the `topic` metadata column should not be writable, but I think it's
worth saving the processing. Updated
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
