AVRO模式更新的问题

Question

我有一个简单的案例类：

  case class User(id: String, login: String, key: String)

我正在添加字段“name”

  case class User(id: String, login: String, name: String, key: String)

{
  "namespace": "test",
  "type": "record",
  "name": "User",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "login", "type": "string" },
    { "name": "name", "type": "string" },
    { "name": "key", "type": "string" }
  ]
}

case class AuthRequest(user: User, session: String)

{
  "namespace": "test",
  "type": "record",
  "name": "AuthRequest",
  "fields": [
    { "name": "user", "type": "User" },
    { "name": "session", "type": "string" }
  ]
}

Consumer.committableSource(consumerSettings, Subscriptions.topics("token_service_auth_request"))
    .map { msg =>
      Try {
        val in: ByteArrayInputStream = new ByteArrayInputStream(msg.record.value())
        val input: AvroBinaryInputStream[AuthRequest] = AvroInputStream.binary[AuthRequest](in)
        val result: AuthRequest = input.iterator.toSeq.head !!!! here is exception
        msg.committableOffset.commitScaladsl()

        (msg.record.value(), result, msg.record.key())
      } match {
        case Success((a: Array[Byte], value: AuthRequest, key: String)) =>
          log.info(s"listener got $msg -> $a -> $value")

          context.parent ! value

        case Failure(e) => e.printStackTrace()
      }
    }
    .runWith(Sink.ignore)

java.util.NoSuchelementException:scala.collection.immutable.stream$empt$.head(stream.scala:1104)在scala.collection.immutable.stream$empt$.head(stream.scala:1102)在test.consumers.authRequestListener.$anonfun$new$2(AuthRequestListener.scala:39)在scala.util.try$.apply(Try.scala:209)在ka.stream.impl.fusing.map$$Anon$9.在Akka.stream.impl.fusing.graphinterpush（ops.scala:51)在Akka.stream.impl.fusing.graphinterpush（graphinterparter.scala:519)在Akka.stream.impl.fusing.graphinterpush（graphinterparter.scala:482)，在Akka.stream.impl.fusing.graphinterpash事件（graphinterparter.scala:488)，在在Akka.stream.impl.fusing.actorGraphGraphGraphGraphGraphGraphGraphGraphGrapperter.Akka$stream$impl$fusing$ActorGraphGrapperInterper$$ProcessEvent(ActorGraphGrapperInterper.scala:563)在Akka.stream.impl.fusing.Acka$stream$impl$fusing$ActorGraphGraphInterperer$$processEvent在Akka.actor.actorcell.receiveMessage（actorcell.scala:588)在Akka.actor.actorcell.invoke（actorcell.scala:557)在Akka.dispatch.mailbox.processMailbox（mailbox.scala:258)在Akka.dispatch.mailbox.run（mailbox.scala:225)在Akka.dispatch.forkjoinask.doexec（mailbox.scala:235)在java:1339)位于akka.dispatch.forkjoin.forkjoinpool.runworker（forkjoinpool.java：1979）位于Akka.dispatch.forkjoin.forkjoinworkerthread.run（forkjoinworkerthread.java：107）

我试图清理构建和无效缓存-我似乎像某种缓存以前版本的模式在一些有帮助请！

Answer 1

您需要使更改向后兼容，使新字段为空，并向其添加默认值。

{
  "namespace": "test",
  "type": "record",
  "name": "User",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "login", "type": "string" },
    { "name": "name", "type": ["null", "string"], "default": null },
    { "name": "key", "type": "string" }
  ]
}