Как использовать zio kafka с google protobuf, когда вам нужно прочитать данные из темы и получить их как прототип класса Java?

Question

Мне нужно получить данные из темы Kafka в виде Zio Stream, данные есть в формате googleprotobuf, также мне нужно проверить схему

Я использую следующий пример файла protobuf, который создает для меня класс proto.Data Java:

      syntax = "proto3";
package proto;

import "google/protobuf/timestamp.proto";

option java_multiple_files = true;
option java_outer_classname = "Protos";

message Data {
  string id = 1;
  google.protobuf.Timestamp receiveTimestamp = 2;
}

Если я использую следующие свойства, я могу получить данные как KStream[proto.Data] (таким образом, используя kafka api) для класса proto.Data proto Message

        val props: Properties = {
    val p = new Properties()
    p.put(StreamsConfig.APPLICATION_ID_CONFIG, s"kstream-application-${java.util.UUID.randomUUID().toString}")
    p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    p.put("security.protocol", "SSL")
    p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName)
    p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, "io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde")
    p.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081")
    p.put("enable.auto.commit", "false")
    p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    p.put("specific.protobuf.value.type", classOf[proto.Data])
    p
  }

А вот пример кода с использованием KStream (я могу распечатать запись только с точным идентификатором, равным 1):

        val builder: StreamsBuilder = new StreamsBuilder
  val risks: KStream[String, proto.Data] =
    builder
      .stream[String, proto.Data](topic)
      .filter((_, value) => value.getId=="1")

  val sysout = Printed
    .toSysOut[String, proto.Data]
    .withLabel("protoStream")
  risks.print(sysout)
  val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
  streams.start()

  sys.ShutdownHookThread {
    streams.close(Duration.ofSeconds(10))
  }

Теперь, если я использую zio kafka и те же свойства, я могу распечатать весь поток:

        val props: Map[String, AnyRef] = Map(
    StreamsConfig.APPLICATION_ID_CONFIG -> s"kstream-application-${java.util.UUID.randomUUID().toString}",
    StreamsConfig.BOOTSTRAP_SERVERS_CONFIG ->  "localhost:9092",
    "security.protocol" -> "SSL",
    StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG ->  Serdes.String.getClass.getName,
    StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG ->  "io.confluent.kafka.streams.serdes.protobuf.KafkaProtobufSerde",
    AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG ->  "http://localhost:8081",
    "enable.auto.commit" ->  "false",
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG ->  "earliest",
    "specific.protobuf.value.type" ->  classOf[proto.Data]
  )

  val myStream = for {

    serdeProto <- Serde.fromKafkaSerde(new KafkaProtobufSerde[proto.Data](), props, true)
    _ <- stream
      .plainStream(Serde.string, serdeProto)
      .provideSomeLayer(consumer ++ Console.live)
      .tap(r => console.putStrLn(s"stream: $r"))
      .runDrain
  } yield ()


  override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] = {
    myStream.exitCode
  }

Но если я попытаюсь отфильтровать только запись с идентификатором, равным 1

        val myStream = for {

    serdeProto <- Serde.fromKafkaSerde(new KafkaProtobufSerde[proto.Data](), props, true)
    _ <- stream
      .plainStream(Serde.string, serdeProto)
      .provideSomeLayer(consumer ++ Console.live)
      .filter(_.value.getId=="1")
      .tap(r => console.putStrLn(s"stream: $r"))
      .runDrain
  } yield ()

Я получаю ошибку, как

      Fiber failed.
An unchecked error was produced.
java.lang.ClassCastException: com.google.protobuf.DynamicMessage cannot be cast to proto.Data

Мне было интересно, использовал ли кто-нибудь zio kafka вместе с googleprotobuf, и была ли десериализация в класс прототипа Java успешной, когда вы читаете данные из темы?