AWSKafkaAvroSerDe is clunky to use, create GenericRecordAWSKafkaAvroSerDe and SpecificRecordAWSKafkaAvroSerDe

[er1c]

(Note: this is some scala code)

But if I have some scala code like:

  val stream = builder.stream(topicName)(
    Consumed.`with`(keySerde, valueSerde)
  ).mapValues { v =>
    val rec = v.asInstanceOf[GenericRecord]

It requires me to do an explicit type cast since AWSKafkaAvroSerDe implements Serde[Object]

It would be much better ergonomics to have something like:

package foo

import com.amazonaws.services.schemaregistry.kafkastreams.AWSKafkaAvroSerDe
import com.amazonaws.services.schemaregistry.utils.{AWSSchemaRegistryConstants, AvroRecordType}
import org.apache.avro.generic.GenericRecord
import org.apache.kafka.common.serialization.{Deserializer, Serde, Serializer}
import software.amazon.awssdk.services.glue.model.DataFormat

import scala.collection.JavaConverters._

object GenericRecordAWSKafkaAvroSerDe {
  private val formatProps = Map(
    AWSSchemaRegistryConstants.AVRO_RECORD_TYPE -> AvroRecordType.GENERIC_RECORD.getName,
    AWSSchemaRegistryConstants.DATA_FORMAT -> DataFormat.AVRO.name(),
  )
}

class GenericRecordAWSKafkaAvroSerDe extends Serde[GenericRecord] {
  private val inner: AWSKafkaAvroSerDe = new AWSKafkaAvroSerDe()

  override def serializer(): Serializer[GenericRecord] =
    inner.serializer().asInstanceOf[Serializer[GenericRecord]]

  override def deserializer(): Deserializer[GenericRecord] =
    inner.deserializer().asInstanceOf[Deserializer[GenericRecord]]

  /**
   * Configure the serializer and de-serializer wrapper.
   *
   * @param serdeConfig          configuration elements for the wrapper
   * @param isSerdeForRecordKeys true if key, false otherwise
   */
  final def configure(serdeConfig: Map[String, _], isSerdeForRecordKeys: Boolean): Unit = {
    val updatedConfig = GenericRecordAWSKafkaAvroSerDe.formatProps ++ serdeConfig
    inner.serializer.configure(updatedConfig.asJava, isSerdeForRecordKeys)
    inner.deserializer.configure(updatedConfig.asJava, isSerdeForRecordKeys)
  }
}

and a corresponding one for SpecificRecord.