Flink AvroDeserializationSchema for SpecificRecord -- Usage with Generic Class Type

Question

Goal: I want to implement a function that can be reused to create the AvroDeserializationSchema for specific records based on the generated Classes that are passed as Type Param. The generated classes are generated using AvroHugger and inherit SpecificRecordBase.

Problem: Unable to find a way on setting up the generic type params to make the compilation work. The AvroDeserializationSchema.forSpecific needs to get a class as argument that was generated.

Here is the function:

  ConfluentRegistryAvroDeserializationSchema[T] = {
    ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classOf[T],
      schemaRegistryUrl
    )
  }

Here is the usage:

val registry = AvroSerde.createEventDeserializerForRegistry[SomeGeneratedClass]("localhost")

Here is the error I get when running the code:

class type required but T found classOf[T]

I know that classOf will not work with arbitrary types (and T is an arbitrary type), I do not have an idea though how I could pass a Class here so that I can reuse this function to pass an arbitrary generated class.

So if I try it with passing in the runtime class of T it does not work either:

ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classTag[T].runtimeClass,
      schemaRegistryUrl
    )

The Error:

cannot be applied to (Class[_$1], String) ConfluentRegistryAvroDeserializationSchema.forSpecific(

If anyone has an idea of how I can implement this abstraction I would be very grateful.

Thanks.

score 0 · Answer 1 · answered Sep 10 '20 at 20:33

I'm not the biggest scala guy in the world, but I'd assume that you need to specify the lower bound for your type parameter (and this is actually a scala question).

You can see in the Java signature that T must extend SpecificRecord

    public static <T extends SpecificRecord> ConfluentRegistryAvroDeserializationSchema<T> forSpecific(Class<T> tClass, String url)

So your function should also specify the same lower type

  ConfluentRegistryAvroDeserializationSchema[T >: SpecificRecord] = {
    ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classOf[T],
      schemaRegistryUrl
    )
  }

Also you probably also want to check how scala reifies the generic types, for example, here.

Flink AvroDeserializationSchema for SpecificRecord -- Usage with Generic Class Type

1 Answers1