[FLINK-18075][hotfix] Use DeserializationSchema instead of KeyedDeser…

…ializationSchema in DynamicTableSink
is · Jun 8, 2020 · 87d6a76 · 87d6a76
1 parent 50b6d9d
commit 87d6a76
Show file tree

Hide file tree

Showing 4 changed files with 4 additions and 8 deletions.
diff --git a/...fka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java b/...fka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java
@@ -21,7 +21,6 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.types.Row;
@@ -57,7 +56,7 @@ protected SinkFunction<Row> createKafkaProducer(
  Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer011<>(
  topic,
- new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ serializationSchema,
  properties,
  partitioner);
  }

diff --git a/.../src/main/java/org/apache/flink/streaming/connectors/kafka/table/Kafka011DynamicSink.java b/.../src/main/java/org/apache/flink/streaming/connectors/kafka/table/Kafka011DynamicSink.java
@@ -22,7 +22,6 @@
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
-import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.connector.format.EncodingFormat;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
@@ -60,7 +59,7 @@ protected SinkFunction<RowData> createKafkaProducer(
  Optional<FlinkKafkaPartitioner<RowData>> partitioner) {
  return new FlinkKafkaProducer011<>(
  topic,
- new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ serializationSchema,
  properties,
  partitioner);
  }

diff --git a/...ector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java b/...ector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java
@@ -20,7 +20,6 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.types.Row;
@@ -52,7 +51,7 @@ protected SinkFunction<Row> createKafkaProducer(
  Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer<>(
  topic,
- new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ serializationSchema,
  properties,
  partitioner);
  }

diff --git a/...fka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java b/...fka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
@@ -21,7 +21,6 @@
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
-import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.connector.format.EncodingFormat;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
@@ -59,7 +58,7 @@ protected SinkFunction<RowData> createKafkaProducer(
  Optional<FlinkKafkaPartitioner<RowData>> partitioner) {
  return new FlinkKafkaProducer<>(
  topic,
- new KeyedSerializationSchemaWrapper<>(serializationSchema),
+ serializationSchema,
  properties,
  partitioner);
  }