Looking for some help with BytesMessageProtoMapper.

[rahul-nu]

I am trying to follow along with the kafka example in the source code. The following works for me, but is slower than deserializing json strings with orjson:

  schema_client = SchemaRegistryClient({"url": "http://redpanda.internal:8081"})
  key_deser = ProtobufDeserializer(
      trade_key_pb2.Base, schema_registry_client=schema_client
  )
  value_deser = ProtobufDeserializer(
      trade_value_pb2.Trade, schema_registry_client=schema_client
  )
  
  
  @csp.node
  def deserialize(
      datum: ts[bytes],
  ) -> ts[tuple[trade_key_pb2.Base, trade_value_pb2.Trade]]:
      key = key_deser(datum, SerializationContext("trade", MessageField.KEY))
      trade = value_deser(datum, SerializationContext("trade", MessageField.VALUE))
      return key, trade

    kafka = KafkaAdapterManager(broker)
    data = kafka.subscribe(
        ts_type=bytes,
        msg_mapper=RawBytesMessageMapper(),
        topic='trade',
    )
   csp.print("msg", deserialize(data))

With appropriately defined class ProtoMessage(csp.Struct), I tried

    msg_mapper = BytesMessageProtoMapper(
        proto_directory="proto/",
        proto_filename="trade-key.proto",
        proto_message="Snapshot",
    )
    data = kafka.subscribe(
        ts_type=ProtoMessage,
        msg_mapper = msg_mapper,
        topic=topic,
    )

which leads to
ValueError: MessageStructConverter.cpp:create:51:ValueError: MessageStructConverter for type PROTOBUF is not defined
Also, what is proto_message, and how does it relate to the (de)serialization?
Is there a solution where the topic-key and topic-value are in different .proto files?