dmariassy commented on code in PR #109:
URL:
https://github.com/apache/flink-connector-kafka/pull/109#discussion_r1690320572
##########
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:
> then for topics.size == 1 the topic metadata column should not be writable
I think that makes a lot of sense!
--
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
