[FLINK-9979] [table] Support a FlinkKafkaPartitioner for Kafka table …

…sink factory

Adds the possibility to add a FlinkKafkaPartitioner to a Kafka table sink
factory. It povides shortcuts for the built-in "fixed" and "round-robin" partitioning.

This closes apache#6440.

flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSink.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.descriptors.ConnectorDescriptor;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -73,16 +74,23 @@ public Kafka010JsonTableSink(String topic, Properties properties, FlinkKafkaPart
  }
 
  @Override
- protected FlinkKafkaProducerBase<Row> createKafkaProducer(String topic, Properties properties, SerializationSchema<Row> serializationSchema, FlinkKafkaPartitioner<Row> partitioner) {
+ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
+ String topic,
+ Properties properties,
+ SerializationSchema<Row> serializationSchema,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer010<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 
  @Override
  protected Kafka010JsonTableSink createCopy() {
- return new Kafka010JsonTableSink(topic, properties, partitioner);
+ return new Kafka010JsonTableSink(
+ topic,
+ properties,
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 }

flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSink.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -36,7 +37,7 @@ public Kafka010TableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
  super(
  schema,
@@ -51,11 +52,11 @@ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
  String topic,
  Properties properties,
  SerializationSchema<Row> serializationSchema,
- FlinkKafkaPartitioner<Row> partitioner) {
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer010<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(null));
  }
 }

flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactory.java

@@ -77,7 +77,7 @@ protected KafkaTableSink createKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka010TableSink(

flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java

@@ -40,7 +40,10 @@ protected KafkaTableSink createTableSink(
  Properties properties,
  FlinkKafkaPartitioner<Row> partitioner) {
 
- return new Kafka010JsonTableSink(topic, properties, partitioner);
+ return new Kafka010JsonTableSink(
+ topic,
+ properties,
+ partitioner);
  }
 
  @Override

flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSourceSinkFactoryTest.java

@@ -85,7 +85,7 @@ protected KafkaTableSink getExpectedKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka010TableSink(

flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSink.java

@@ -39,7 +39,7 @@ public Kafka011TableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
  super(
  schema,
@@ -54,11 +54,11 @@ protected SinkFunction<Row> createKafkaProducer(
  String topic,
  Properties properties,
  SerializationSchema<Row> serializationSchema,
- FlinkKafkaPartitioner<Row> partitioner) {
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer011<>(
  topic,
  new KeyedSerializationSchemaWrapper<>(serializationSchema),
  properties,
- Optional.of(partitioner));
+ partitioner);
  }
 }

flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactory.java

@@ -77,7 +77,7 @@ protected KafkaTableSink createKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka011TableSink(

flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java

@@ -85,7 +85,7 @@ protected KafkaTableSink getExpectedKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka011TableSink(

flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java

@@ -26,6 +26,7 @@
 import org.apache.flink.table.descriptors.ConnectorDescriptor;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -92,17 +93,24 @@ public Kafka08JsonTableSink(String topic, Properties properties, KafkaPartitione
  }
 
  @Override
- protected FlinkKafkaProducerBase<Row> createKafkaProducer(String topic, Properties properties, SerializationSchema<Row> serializationSchema, FlinkKafkaPartitioner<Row> partitioner) {
+ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
+ String topic,
+ Properties properties,
+ SerializationSchema<Row> serializationSchema,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer08<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 
  @Override
  protected Kafka08JsonTableSink createCopy() {
- return new Kafka08JsonTableSink(topic, properties, partitioner);
+ return new Kafka08JsonTableSink(
+ topic,
+ properties,
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 }
 

flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSink.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -36,7 +37,7 @@ public Kafka08TableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
  super(
  schema,
@@ -51,11 +52,11 @@ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
  String topic,
  Properties properties,
  SerializationSchema<Row> serializationSchema,
- FlinkKafkaPartitioner<Row> partitioner) {
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer08<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(null));
  }
 }

flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactory.java

@@ -77,7 +77,7 @@ protected KafkaTableSink createKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka08TableSink(

flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java

@@ -40,7 +40,10 @@ protected KafkaTableSink createTableSink(
  Properties properties,
  FlinkKafkaPartitioner<Row> partitioner) {
 
- return new Kafka08JsonTableSink(topic, properties, partitioner);
+ return new Kafka08JsonTableSink(
+ topic,
+ properties,
+ partitioner);
  }
 
  @Override

flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSourceSinkFactoryTest.java

@@ -85,7 +85,7 @@ protected KafkaTableSink getExpectedKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka08TableSink(

flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSink.java

@@ -26,6 +26,7 @@
 import org.apache.flink.table.descriptors.ConnectorDescriptor;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -92,16 +93,23 @@ public Kafka09JsonTableSink(String topic, Properties properties, KafkaPartitione
  }
 
  @Override
- protected FlinkKafkaProducerBase<Row> createKafkaProducer(String topic, Properties properties, SerializationSchema<Row> serializationSchema, FlinkKafkaPartitioner<Row> partitioner) {
+ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
+ String topic,
+ Properties properties,
+ SerializationSchema<Row> serializationSchema,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer09<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 
  @Override
  protected Kafka09JsonTableSink createCopy() {
- return new Kafka09JsonTableSink(topic, properties, partitioner);
+ return new Kafka09JsonTableSink(
+ topic,
+ properties,
+ partitioner.orElse(new FlinkFixedPartitioner<>()));
  }
 }

flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSink.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.types.Row;
 
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -36,7 +37,7 @@ public Kafka09TableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
  super(
  schema,
@@ -51,11 +52,11 @@ protected FlinkKafkaProducerBase<Row> createKafkaProducer(
  String topic,
  Properties properties,
  SerializationSchema<Row> serializationSchema,
- FlinkKafkaPartitioner<Row> partitioner) {
+ Optional<FlinkKafkaPartitioner<Row>> partitioner) {
  return new FlinkKafkaProducer09<>(
  topic,
  serializationSchema,
  properties,
- partitioner);
+ partitioner.orElse(null));
  }
 }

flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactory.java

@@ -77,7 +77,7 @@ protected KafkaTableSink createKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka09TableSink(

flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java

@@ -40,7 +40,10 @@ protected KafkaTableSink createTableSink(
  Properties properties,
  FlinkKafkaPartitioner<Row> partitioner) {
 
- return new Kafka09JsonTableSink(topic, properties, partitioner);
+ return new Kafka09JsonTableSink(
+ topic,
+ properties,
+ partitioner);
  }
 
  @Override

flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09TableSourceSinkFactoryTest.java

@@ -85,7 +85,7 @@ protected KafkaTableSink getExpectedKafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
 
  return new Kafka09TableSink(

flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSink.java

@@ -40,7 +40,7 @@
  * A version-agnostic Kafka {@link AppendStreamTableSink}.
  *
  * <p>The version-specific Kafka consumers need to extend this class and
- * override {@link #createKafkaProducer(String, Properties, SerializationSchema, FlinkKafkaPartitioner)}}.
+ * override {@link #createKafkaProducer(String, Properties, SerializationSchema, Optional)}}.
  */
 @Internal
 public abstract class KafkaTableSink implements AppendStreamTableSink<Row> {
@@ -60,7 +60,7 @@ public abstract class KafkaTableSink implements AppendStreamTableSink<Row> {
  protected Optional<SerializationSchema<Row>> serializationSchema;
 
  /** Partitioner to select Kafka partition for each item. */
- protected final FlinkKafkaPartitioner<Row> partitioner;
+ protected final Optional<FlinkKafkaPartitioner<Row>> partitioner;
 
  // legacy variables
  protected String[] fieldNames;
@@ -70,7 +70,7 @@ protected KafkaTableSink(
  TableSchema schema,
  String topic,
  Properties properties,
- FlinkKafkaPartitioner<Row> partitioner,
+ Optional<FlinkKafkaPartitioner<Row>> partitioner,
  SerializationSchema<Row> serializationSchema) {
  this.schema = Optional.of(Preconditions.checkNotNull(schema, "Schema must not be null."));
  this.topic = Preconditions.checkNotNull(topic, "Topic must not be null.");
@@ -96,7 +96,7 @@ public KafkaTableSink(
  this.schema = Optional.empty();
  this.topic = Preconditions.checkNotNull(topic, "topic");
  this.properties = Preconditions.checkNotNull(properties, "properties");
- this.partitioner = Preconditions.checkNotNull(partitioner, "partitioner");
+ this.partitioner = Optional.of(Preconditions.checkNotNull(partitioner, "partitioner"));
  this.serializationSchema = Optional.empty();
  }
 
@@ -113,7 +113,7 @@ protected abstract SinkFunction<Row> createKafkaProducer(
  String topic,
  Properties properties,
  SerializationSchema<Row> serializationSchema,
- FlinkKafkaPartitioner<Row> partitioner);
+ Optional<FlinkKafkaPartitioner<Row>> partitioner);
 
  /**
  * Create serialization schema for converting table rows into bytes.