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?

[robambalu]

Apologies for the delayed response here. At the moment the adapter doesnt support protobuf keys, only data.
As for your node, im not sure how expensive it is to create those SerializationContext objects, but if they can be reused then I would suggest storing them in node state variables so they are only constructed once at startup
https://github.com/Point72/csp/wiki/CSP-Node#anatomy-of-a-cspnode