flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] tragicjun closed pull request #6495: [FLINK-10061] [table] [kafka] Fix unsupported reconfiguration in KafkaTableSink
Date Tue, 28 Aug 2018 08:38:50 GMT
tragicjun closed pull request #6495: [FLINK-10061] [table] [kafka] Fix unsupported reconfiguration
in KafkaTableSink
URL: https://github.com/apache/flink/pull/6495

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
index a85d536eac9..772a94a2c4f 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
@@ -26,6 +26,7 @@
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.util.TableConnectorUtil;
 import org.apache.flink.types.Row;
@@ -166,8 +167,15 @@ public void emitDataStream(DataStream<Row> dataStream) {
 	public KafkaTableSink configure(String[] fieldNames, TypeInformation<?>[] fieldTypes)
 		if (schema.isPresent()) {
-			// a fixed schema is defined so reconfiguration is not supported
-			throw new UnsupportedOperationException("Reconfiguration of this sink is not supported.");
+			if (getFieldTypes().length != fieldTypes.length || !Arrays.deepEquals(getFieldTypes(),
fieldTypes)) {
+				String fixedFieldTypes = Arrays.toString(getFieldTypes());
+				String configuredFieldTypes = Arrays.toString(fieldTypes);
+				throw new ValidationException("A fixed table schema is defined but the configured fieldTypes
" +
+						configuredFieldTypes + " does not match with the fieldTypes " + fixedFieldTypes +
+						" declared by the fixed table schema.");
+			}
+			return this;
 		// legacy code


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:

With regards,
Apache Git Services

View raw message