question re IllegalArgumentException: Unsupported Avro type. Supported types are null, Boolean, Integer, Long, Float, Double, String, byte[] and IndexedRecord

[nicolaemarasoiu]

Hello,

We have like this:

addSbtPlugin("com.julianpeeters" % "sbt-avrohugger" % "2.0.0-RC22")
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.5")
addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.0.2")
addSbtPlugin("org.wartremover" % "sbt-wartremover" % "2.4.3")

scalaVersion := "2.12.10"
name := "global-topic-conveyor"
organization := "com.ovoenergy.data-availability"
version := "1.0"
scalacOptions ++= Seq(
"-deprecation",
"-feature",
"-unchecked",
"-language:higherKinds",
// "-Xfatal-warnings",
"-Xlint:infer-any",
"-Xlint:unused"
)

resolvers += "confluent" at "https://packages.confluent.io/maven/"

val confluentVersion = "5.4.0"
val kafkaVersion = confluentVersion + "-ccs"

libraryDependencies += "org.apache.kafka" %% "kafka-streams-scala" % kafkaVersion
libraryDependencies += "io.confluent" % "kafka-streams-avro-serde" % confluentVersion
libraryDependencies += "io.confluent" % "kafka-schema-registry-client" % confluentVersion
libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.2.3"
libraryDependencies += "com.typesafe" % "config" % "1.4.0"
libraryDependencies += "com.sksamuel.avro4s" %% "avro4s-core" % "3.0.4"
libraryDependencies += "io.circe" %% "circe-parser" % "0.13.0"
libraryDependencies += "org.apache.kafka" % "kafka-streams-test-utils" % kafkaVersion % "test"
libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.8" % "test"
libraryDependencies += "com.github.tomakehurst" % "wiremock" % "2.25.1" % "test"

avroSourceDirectories in Compile += (sourceDirectory in Compile).value / "resources" / "avro"
sourceGenerators in Compile += (avroScalaGenerate in Compile).taskValue

assemblyJarName in assembly := "global-topic-conveyor.jar"

parallelExecution in Test := false

sourceGenerators in Compile += Def.task {
val schemasPath = (sourceDirectory in Compile).value / "resources" / "avro"
streams.value.log.info(s"Parsing schemas from $schemasPath")
val managedFolder = (sourceManaged in Compile).value
val outFile = CreateImplicitsForSchemas.writeImplicitsForSchemas(schemasPath, managedFolder)
streams.value.log.info(s"Generated implicits for Avro4s: $outFile")
Seq(outFile)
}.taskValue

wartremoverErrors in (Compile, compile) ++= Warts.unsafe.filterNot(_ == Wart.Any) ++ Seq(
Wart.Nothing
)

wartremoverExcluded += sourceManaged.value
assemblyMergeStrategy in assembly := {
// TODO: module-info.class is conflicting for com.fasterxml.jackson after importing the new kafka libraries
// This has been addressed by discarding the file from the fat jar (as per the MergeStrategy.discard)
case "module-info.class" => MergeStrategy.discard
case x =>
val oldStrategy = (assemblyMergeStrategy in assembly).value
oldStrategy(x)
}

// Since the uberjar is deployed by the CI process and tests are performed before the jar is built
// There's no need for testing again before assembling, hence the next line
test in assembly := {}

package com.ovoenergy.globaltopics.pipelines.orion

import java.time.{Instant, LocalDate}
import java.util.{Properties, UUID}

import com.ovoenergy.kafka.common.event.Metadata
import com.ovoenergy.kafka.identity.event.{
CreatedOrUpdate,
Deleted,
FormattedAddress,
User,
UserEvent
}
import io.confluent.kafka.serializers.{AbstractKafkaAvroSerDeConfig, KafkaAvroSerializer}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

object Producer {
def main(args: Array[String]): Unit = {
val topic = "identity_users_v1"
val props = fromMap(
Map(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> "localhost:9092",
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> classOf[StringSerializer],
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> classOf[KafkaAvroSerializer],
AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> "http://localhost:8081"
)
)
val producer = new KafkaProducerString, UserEvent
val userEventId = "userEventId__alsjbfahjlsbefhjashdfbjsaf"
val userId = "userId__qowiuefg324uf3qhf3q4"
val inputMessageValue = UserEvent(
metadata = Metadata(eventId = userEventId, traceToken = "asdfkjasdf", createdAt = Instant.now),
realm = "ovo-france",
event = Left[CreatedOrUpdate, Deleted](
CreatedOrUpdate(
User(
id = userId,
title = Some("Mr"),
givenName = "John",
familyName = "Smith",
emailAddress = Some("[email protected]"),
birthdate = Some(LocalDate.of(1978, 6, 4)),
phoneNumbers = List("072189736"),
permissions = Nil,
postalAddress = Some(List("Unused address, we use the formattedAddresses instead")),
formattedAddresses = List(
FormattedAddress(
addressType = "current",
country = "GB",
address = """
{
"line1":"195 Radburn Close",
"town":"Eastbourne",
"postcode":"CM18 7EQ"
}
"""
)
)
)
)
)
)
discard(
producer.send(
new ProducerRecord[String, UserEvent](topic, UUID.randomUUID.toString, inputMessageValue)
)
)
producer.flush()
}
def discard(evaluateForSideEffectOnly: Any): Unit = {
val _: Any = evaluateForSideEffectOnly
()
}

def fromMap(properties: Map[String, Object]): Properties = {
val result = properties.foldLeft(new Properties()) {
case (props, (k, v)) =>
discard(props.put(k, v))
props
}
println("For map=" + properties.toString() + ", properties=" + result.toString)
result
}

}

log:

{"message":"[Producer clientId=producer-1] Cluster ID: sqjPEJjHQyuly4zXPuqdmA","logger":"org.apache.kafka.clients.Metadata","thread":"kafka-producer-network-thread | producer-1","severity":"INFO","timestamp":{"seconds":1587741874,"nanos":201000000}}

Exception in thread "main" java.lang.IllegalArgumentException: Unsupported Avro type. Supported types are null, Boolean, Integer, Long, Float, Double, String, byte[] and IndexedRecord
at io.confluent.kafka.serializers.AvroSchemaUtils.getSchema(AvroSchemaUtils.java:93)
at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:55)
at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:903)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:865)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:752)
at com.ovoenergy.globaltopics.pipelines.orion.Producer$.main(Producer.scala:66)
at com.ovoenergy.globaltopics.pipelines.orion.Producer.main(Producer.scala)