klam-shop commented on code in PR #109:
URL:
https://github.com/apache/flink-connector-kafka/pull/109#discussion_r1688236979
##########
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);
+ } else if (topics != null && topics.size() > 0 &&
!topics.contains(topic)) {
Review Comment:
Open to discuss: I decided to keep topics as a List since, but we can change
it to HashSet. I am assuming the list will be short on average.
List can outperform HashSet for small number of elements:
https://stackoverflow.com/questions/150750/hashset-vs-list-performance
--
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
