我有一个简单的案例类:
case class User(id: String, login: String, key: String)
我添加字段“名称”
case class User(id: String, login: String, name: String, key: String)
然后在avro架构中添加此字段(user.avsc)
{
"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)
chema(auth_request.avsc)
{
"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 $ empty $ .head(Stream.scala:1104)at scala.collection.immutable.Stream $ empty $ .head(Stream.scala:1102) )在test.consumers.AuthRequestListener . $ anonfun $ new $ 2(AuthRequestListener.scala:39)at scala.util.Try $ .apply(Try.scala:209)at test.consumers.AuthRequestListener . $ anonfun $ new $ 1(AuthRequestListener) .scala:36)在test.consumers.AuthRequestListener . $ anonfun $ new $ 1 $ adapted(AuthRequestListener.scala:35)at akka.stream.impl.fusing.Map $$ anon $ 9.onPush(Ops.scala:51)at at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:519)at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:482)at akka.stream.impl.fusing.GraphInterpreter.execute( GraphInterpreter.scala:378)akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:588)at akka.stream.impl.fusing.GraphInterpreterShell $ AsyncInput.execute(ActorGraphInterpreter.scala:472)at akka.s trek.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:563)akka.stream.impl.fusing.ActorGraphInterpreter.akka $ stream $ impl $ fusing $ actorGraphInterpreter $$ processEvent(ActorGraphInterpreter.scala:745)at akka.stream .impl.fusing.ActorGraphInterpreter $$ anonfun $在akka.actor.Actor.aroundReceive(Actor.scala:517)akka.actor.Actor.aroundReceive $(Actor.scala:)收到$ 1.applyOrElse(ActorGraphInterpreter.scala:760) 515)akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:670)at akka.actor.ActorCell.receiveMessage(ActorCell.scala:588)at akka.actor.ActorCell.invoke(ActorCell.scala:557) )akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)at akka.dispatch.Mailbox.run(Mailbox.scala:225)at akka.dispatch.Mailbox.exec(Mailbox.scala:235)at akka.dispatch .forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)at akka.dispatch.forkjoin.ForkJoinPool $ WorkQueue.runTask(ForkJoinPool.java:1339)at akka.dispatch.forkjoin.ForkJoinPool.run Worker (ForkJoinPool.java:1979)at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
我试图清理构建并使缓存无效 - 我似乎某种缓存以前版本的架构在某些帮助请!
1 回答
您需要使更改向后兼容,使新字段可以为空,并为其添加默认值 .