add isKeyed to KafkaSinkConfig

epiphanous · Sep 29, 2020 · 74a4564 · 74a4564
1 parent 3ed3aec
commit 74a4564
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/src/main/scala/io/epiphanous/flinkrunner/model/SinkConfig.scala b/src/main/scala/io/epiphanous/flinkrunner/model/SinkConfig.scala
@@ -18,7 +18,11 @@ object SinkConfig {
       case Some(connector) =>
         connector match {
           case Kafka =>
-            KafkaSinkConfig(connector, name, config.getString(s"$p.topic"), config.getProperties(s"$p.config"))
+            KafkaSinkConfig(connector,
+                            name,
+                            config.getString(s"$p.topic"),
+                            config.getBoolean(s"$p.isKeyed"),
+                            config.getProperties(s"$p.config"))
           case Kinesis =>
             KinesisSinkConfig(connector, name, config.getString(s"$p.stream"), config.getProperties(s"$p.config"))
           case File =>
@@ -57,6 +61,7 @@ final case class KafkaSinkConfig(
   connector: FlinkConnectorName = Kafka,
   name: String,
   topic: String,
+  isKeyed: Boolean,
   properties: Properties)
     extends SinkConfig
 final case class KinesisSinkConfig(