ClassTag is provided in params in deserialization schema factory … #1499

gskrobisz · 2021-03-31T10:25:46Z

ClassTag is provided in params so one deserialization schema factory will work for multiple deserialization schemas

...link/avro-util/src/main/scala/pl/touk/nussknacker/engine/avro/KafkaAvroBaseTransformer.scala

arkadius

Good change! It need some slight modifications but looks promising.

...-util/src/main/scala/pl/touk/nussknacker/engine/avro/source/BaseKafkaAvroSourceFactory.scala

arkadius · 2021-03-31T14:29:41Z

...-util/src/main/scala/pl/touk/nussknacker/engine/avro/source/BaseKafkaAvroSourceFactory.scala

+    val valueSchemaData = valueSchemaDeterminer.determineSchemaUsedInTyping.valueOr(SchemaDeterminerErrorHandler.handleSchemaRegistryErrorAndThrowException)
+    val valueSchemaUsedInRuntime = valueSchemaDeterminer.toRuntimeSchema(valueSchemaData)
+    // key schema, optional
+    val keySchemaDataUsedInRuntime = Option(keySchemaDeterminer).flatMap(determiner => {


Use flatMap { determiner => ... instead of flatMap(determiner => { to reduce number of braces.

if you see nested flatMap/map inside flatMap you can use for-comprehension it will be more straightforward

Don't handle error as a None. Let's handle it the same way as it is handled in value approach. On the top level (KafkaAvroBaseTransformer maybe?) we should define determiner that should not even ask schema registry for key schema. It should be done only in tests of TupleAvroKeyValueKafkaAvroDeserializerSchemaFactory

Fixed. "Empty" Determiner is provided as FixedNoneSchemaDeterminer.

passing keySchemaDeterminer is ok, but I don't like this getOrElse. Why you pass determiner if you override it's decision? Why not to just throw exception here?

Fixed. There is one schema determiner used in both contextTransformation and implementation. See BasedOnVersionWithFallbackAvroSchemaDeterminer

arkadius · 2021-03-31T14:38:31Z

...-util/src/main/scala/pl/touk/nussknacker/engine/avro/source/BaseKafkaAvroSourceFactory.scala

+                   valueSchemaDeterminer: AvroSchemaDeterminer,
+                   keySchemaDeterminer: AvroSchemaDeterminer,
+                   returnGenericAvroType: Boolean,
+                   valueClassTagOpt: Option[ClassTag[_]] = None,


I'm not a fan of optional parameters in such a low level api. IMO it is only good for some domain level classes where you want to have some fancy factory methods combinations or in tests. In production code it is easy to miss argument in some invocation where it should be passed.

Fixed. Optional ClassTag parameters are removed (replaced by [K: ClassTag, V: ClassTag])

arkadius · 2021-03-31T14:41:32Z

...la/pl/touk/nussknacker/engine/avro/serialization/KafkaAvroDeserializationSchemaFactory.scala

+
+object KafkaAvroKeyValueDeserializationSchemaFactory {
+
+  val fallbackKeyAsStringDeserializer: Deserializer[String] = new Deserializer[String] {


this deserializer is not a fallbck deserializer. It just derialize to string. It is used as a fallback so "fallback" word could be placed in the place of usage.

lest create named class StringDeserializer - it wil help in some debugging

why you haven't used existing org.apache.kafka.common.serialization.StringDeserializer?

Fixed. Object is removed, fallback scenario should be implemented in a custom deerializer. Here, for value-only deserializers keys are ignored, and for key-value deserializers specific schema should be provided (default scenario).

I still see not used fallbackKeyAsStringDeserializer value, why?

Fixed, removed unused object.

.../schemaregistry/confluent/serialization/ConfluentKafkaAvroDeserializationSchemaFactory.scala

arkadius · 2021-03-31T15:06:39Z

...main/scala/pl/touk/nussknacker/engine/avro/source/SpecificRecordKafkaAvroSourceFactory.scala

+      schemaRegistryProvider.deserializationSchemaFactory,
+      schemaRegistryProvider.recordFormatter,
+      valueSchemaDeterminer,
+      keySchemaDeterminer = null,


No no no, we are not using null in scala! Always use Option instead... But in this case IMO simpler would be to create case object FixedNoneSchemaDeterminer. WDYT?

Fixed. Created FixedNoneSchemaDeterminer.

arkadius · 2021-03-31T15:26:27Z

...cker/engine/avro/schemaregistry/confluent/serialization/ConfluentKafkaAvroDeserializer.scala

@@ -34,8 +34,8 @@ class ConfluentKafkaAvroDeserializer[T](kafkaConfig: KafkaConfig, schemaData: Ru
    KryoGenericRecordSchemaIdSerializationSupport.schemaIdSerializationEnabled(kafkaConfig)

  override def deserialize(topic: String, data: Array[Byte]): T = {
-    val record = deserialize(topic, isKey, data, schemaData)
-    record.asInstanceOf[T]
+    val payloadObj = deserialize(topic, isKey, data, schemaData)


It is not always kafka payload. It can be used also for isKey = true.

Neither it is a "record", which is somewhat misleading. Changed name to neutral deserializedData (to match parameter data).

lciolecki

I wonder about that can we right now choose key schema version on source / sink?

...link/avro-util/src/main/scala/pl/touk/nussknacker/engine/avro/KafkaAvroBaseTransformer.scala

...cala/pl/touk/nussknacker/engine/avro/schemaregistry/BasedOnVersionAvroSchemaDeterminer.scala

.../schemaregistry/confluent/serialization/ConfluentKafkaAvroDeserializationSchemaFactory.scala

lciolecki · 2021-04-07T16:59:49Z

...la/pl/touk/nussknacker/engine/avro/serialization/KafkaAvroDeserializationSchemaFactory.scala

+    *           use other deserialization strategy base on it or provide different TypeInformation
+    * @return KafkaDeserializationSchema
+    */
+  def create[K: ClassTag, V: ClassTag](kafkaConfig: KafkaConfig,


Hmm, I don't like this formatting :)

arkadius · 2021-04-13T11:22:22Z

...ne/flink/avro-util/src/main/scala/pl/touk/nussknacker/engine/avro/AvroSchemaDeterminer.scala

@@ -34,3 +35,8 @@ object RuntimeSchemaData {
 }

 class SchemaDeterminerError(message: String, cause: Throwable) extends RuntimeException(message, cause)
+
+case object FixedNoneSchemaDeterminer extends AvroSchemaDeterminer {


I think that we don't need FixedNoneSchemaDeterminer but FixedStringSchemaDeterminer, because at the end this schema will be used to determine type of key. Also I would add here a description that this determiner is because we want to have one generic source that will handle both situations: key with defined schema in registry and key as fixed string type.

Changed to FixedStringSchemaDeterminer (have not removed, it's used in Specific source)

…will work for multiple deserialization schemas

arkadius

Good job, let's merge it

gskrobisz linked an issue Mar 31, 2021 that may be closed by this pull request

Source should be able to control start variables #548

Closed

gskrobisz commented Mar 31, 2021

View reviewed changes

...link/avro-util/src/main/scala/pl/touk/nussknacker/engine/avro/KafkaAvroBaseTransformer.scala Outdated Show resolved Hide resolved

arkadius reviewed Mar 31, 2021

View reviewed changes

lciolecki reviewed Apr 2, 2021

View reviewed changes

lciolecki reviewed Apr 7, 2021

View reviewed changes

arkadius reviewed Apr 13, 2021

View reviewed changes

gskrobisz force-pushed the improvements/kafka_avro_kv_deserializer_classTag_param branch from 1c4f8e9 to 2985580 Compare April 13, 2021 13:01

gskrobisz added 5 commits April 13, 2021 15:03

ClassTag is provided in params so one deserialization schema factory …

e201537

…will work for multiple deserialization schemas

Code review fixes

42e87a0

Code review fixes - code formatting

d903afa

Migration guide.

ee9aea7

Add AvroSchemaDeterminer with fallback to fixed String schema.

2c87706

gskrobisz force-pushed the improvements/kafka_avro_kv_deserializer_classTag_param branch from 2985580 to 2c87706 Compare April 13, 2021 13:04

arkadius approved these changes Apr 13, 2021

View reviewed changes

gskrobisz merged commit b7c0424 into staging Apr 13, 2021

dswiecki deleted the improvements/kafka_avro_kv_deserializer_classTag_param branch July 21, 2021 16:39

snyk-bot mentioned this pull request Sep 21, 2021

[Snyk] Security upgrade jointjs from 3.0.2 to 3.4.2 lciolecki/nussknacker#12

Open

This was referenced Oct 6, 2022

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#18

Open

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#19

Open

[Snyk] Fix for 20 vulnerabilities lciolecki/nussknacker#20

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ClassTag is provided in params in deserialization schema factory … #1499

ClassTag is provided in params in deserialization schema factory … #1499

gskrobisz commented Mar 31, 2021

arkadius left a comment

arkadius Mar 31, 2021

gskrobisz Apr 7, 2021

arkadius Apr 13, 2021

gskrobisz Apr 13, 2021

arkadius Mar 31, 2021

gskrobisz Apr 7, 2021 •

edited

Loading

arkadius Mar 31, 2021

gskrobisz Apr 7, 2021

arkadius Apr 13, 2021

gskrobisz Apr 13, 2021

arkadius Mar 31, 2021

gskrobisz Apr 7, 2021

arkadius Mar 31, 2021

gskrobisz Apr 7, 2021

lciolecki left a comment

lciolecki Apr 7, 2021

gskrobisz Apr 8, 2021

arkadius Apr 13, 2021

gskrobisz Apr 13, 2021

arkadius left a comment


		object KafkaAvroKeyValueDeserializationSchemaFactory {

		val fallbackKeyAsStringDeserializer: Deserializer[String] = new Deserializer[String] {

ClassTag is provided in params in deserialization schema factory … #1499

ClassTag is provided in params in deserialization schema factory … #1499

Conversation

gskrobisz commented Mar 31, 2021

arkadius left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gskrobisz Apr 7, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lciolecki left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arkadius left a comment

Choose a reason for hiding this comment

gskrobisz Apr 7, 2021 •

edited

Loading