Universal kafka components: handling both avro and json - inital work #2

TouK · Jul 11, 2022 · 5fc277d · 5fc277d
1 parent ba25bba
commit 5fc277d
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/...scala/pl/touk/nussknacker/engine/flink/util/transformer/FlinkKafkaComponentProvider.scala b/...scala/pl/touk/nussknacker/engine/flink/util/transformer/FlinkKafkaComponentProvider.scala
@@ -53,8 +53,8 @@ class FlinkKafkaComponentProvider extends ComponentProvider {
 
     // TODO: change link to the documentation when json schema handling will be available
     val universalKafkaComponents = List(
-      ComponentDefinition("kafka", new UniversalKafkaSourceFactory(avroSerializingSchemaRegistryProvider, dependencies, new FlinkKafkaSourceImplFactory(None))).withRelativeDocs(avro),
-      ComponentDefinition("kafka", new UniversalKafkaSinkFactory(avroSerializingSchemaRegistryProvider, dependencies, FlinkKafkaAvroSinkImplFactory)).withRelativeDocs(avro))
+      ComponentDefinition("kafka", new UniversalKafkaSourceFactory(avroSerializingSchemaRegistryProvider, overriddenDependencies, new FlinkKafkaSourceImplFactory(None))).withRelativeDocs(avro),
+      ComponentDefinition("kafka", new UniversalKafkaSinkFactory(avroSerializingSchemaRegistryProvider, overriddenDependencies, FlinkKafkaAvroSinkImplFactory)).withRelativeDocs(avro))
 
     lowLevelKafkaComponents ::: universalKafkaComponents
   }

diff --git a/...rc/main/scala/pl/touk/nussknacker/engine/lite/components/LiteKafkaComponentProvider.scala b/...rc/main/scala/pl/touk/nussknacker/engine/lite/components/LiteKafkaComponentProvider.scala
@@ -72,7 +72,7 @@ class LiteKafkaComponentProvider(schemaRegistryClientFactory: ConfluentSchemaReg
 
     // TODO: change link to the documentation when json schema handling will be available
     val universalKafkaComponents = List(
-      ComponentDefinition(KafkaAvroName, new UniversalKafkaSourceFactory(avroSerializingSchemaRegistryProvider, dependencies, new LiteKafkaSourceImplFactory)).withRelativeDocs(avro),
+      ComponentDefinition(KafkaUniversalName, new UniversalKafkaSourceFactory(avroSerializingSchemaRegistryProvider, dependencies, new LiteKafkaSourceImplFactory)).withRelativeDocs(avro),
       ComponentDefinition(KafkaUniversalName, new UniversalKafkaSinkFactory(avroSerializingSchemaRegistryProvider, dependencies, LiteKafkaAvroSinkImplFactory)).withRelativeDocs(avro))
 
     lowLevelKafkaComponents ::: universalKafkaComponents