I'm trying to use Avro4k with Apache Beam by creating a generic `CustomCoder` implementation that ca...

# serialization

[rocketraman]

I'm trying to use Avro4k with Apache Beam by creating a generic CustomCoder implementation that can handle any @Serializable. Avro4k needs a KSerializer<T>. How can I obtain the KSerializer at runtime? Keep in mind that in Beam Coder needs to be serializable itself, so KSerializer or KType cannot be a property of the coder i.e. this does not work, nor does passing in `KType`:

class Avro4kCoder<T: Any>(private val serializer: KSerializer<T>): CustomCoder<T>() {
  override fun encode(value: T, outStream: OutputStream) {
    Avro.default.openOutputStream(serializer) {
      encodeFormat = AvroEncodeFormat.Binary
    }.to(outStream).write(value).flush()
  }
  override fun decode(inStream: InputStream): T? {
    return Avro.default.openInputStream(serializer) {
      decodeFormat = AvroDecodeFormat.Binary(Avro.default.schema(serializer))
    }.from(inStream).next()
  }
}

[Vampire]

Call .serializer() on it?

[rocketraman]

The problem is that the return value is not itself serializable. Apparently Beam needs coders to be serializable, so you end up with errors like:

Caused by: java.io.NotSerializableException: Foo$$serializer

[rocketraman]

Oh, there is no .serializer() on Class. There is on KClass , but that is an internal API.

[Vampire]

What is internal?
KClass is not internal.
And the serializer methods are generated by the kotlinx.serialization compiler plugin afair

[rocketraman]

KClass.serializer() is internal

[Vampire]

Do you want a serializer for a class for for an instance of a class?

[Vampire]

I don't think the former makes too much sense and the latter is like I said, generated by the compiler plugin afair

[rocketraman]

I want a serializer for an instance. However, I'm trying to create a generic CustomCoder which can handle any type which is @Serializable.

[rocketraman]

This is what the coder looks like now:

class Avro4kCoder<T: Any>(private val serializer: KSerializer<T>): CustomCoder<T>() {
  override fun encode(value: T, outStream: OutputStream) {
    Avro.default.openOutputStream(serializer) {
      encodeFormat = AvroEncodeFormat.Binary
    }.to(outStream).write(value).flush()
  }

  override fun decode(inStream: InputStream): T? {
    return Avro.default.openInputStream(serializer) {
      decodeFormat = AvroDecodeFormat.Binary(Avro.default.schema(serializer))
    }.from(inStream).next()
  }
}

[rocketraman]

But that doesn't work because KSerializer<T> is not serializable, which causes a NotSerializableException at runtime.

[rocketraman]

So I was thinking if I could pass a KClass in the constructor instead, and create the serializer at execution time, it would work. However, as I said, that API is internal.

[rocketraman]

Tried to pass in a KType instead -- looks like the public API allows one to call serializer(KType) to get an instance of KSerializer . Unfortunately, Beam complains at runtime about that too:

Caused by: java.io.NotSerializableException: kotlin.reflect.jvm.internal.KTypeImpl

[rocketraman]

I edited the original question for clarity...

[Gilles Barbier]

To get KSerializer at runtime from a java context, I have done this:

fun getKSerializerOrNull(klass: Class<*>): KSerializer<*>? {
    val companionField = klass.declaredFields.find {
        it.name == "Companion" && isStatic(it.modifiers)
    } ?: return null
    val companion = companionField.get(klass)
    val serializerMethod = try {
        companion::class.java.getMethod("serializer")
    } catch (e: NoSuchMethodException) {
        return null
    }
    if (serializerMethod.returnType.name != KSerializer::class.qualifiedName) {
        return null
    }
    @Suppress("UNCHECKED_CAST")
    return serializerMethod.invoke(companion) as KSerializer<*>
}

[Gilles Barbier]

But I agree it's not satisfying

[Gilles Barbier]

see also https://kotlinlang.slack.com/archives/C7A1U5PTM/p1604597507280200

[Gilles Barbier]

In a similar situation, I also did a simple function

fun <T : Any> kserializer(klass: KClass<T>) = when (klass) {
    MyClass1::class -> MyClass1.serializer()
    MyClass2::class -> MyClass2.serializer()
    MyClass3::class -> MyClass3.serializer()
    else -> throw RuntimeException("This should not happen: apply kserializer with  ${klass.qualifiedName}")
} as KSerializer <T>

as the set of classes being called was limited (not very elegant neither, but at least I do not have any reflection here)