Protobuf

The Protobuf codec converts between binary Protobuf messages and columnar Arrow RecordBatches using a descriptor compiled from .proto files at startup. Decoding parses each byte payload against the configured MessageDescriptor; encoding reverses the process. Use it when an input emits raw Protobuf (no Confluent schema-id prefix).

Configuration

Field	Type	Required	Default	Description
type	string	yes	—	Fixed value "protobuf"
message_type	string	yes	—	Fully qualified Protobuf message type name (including package), e.g. com.example.User
proto_inputs	array<string>	yes	—	List of .proto source file paths
proto_includes	array<string>	no	—	Include search paths used when parsing .proto files
on_error	string	no	fail	Decode-error policy: fail (default) fails the whole batch on the first bad message; skip isolates bad messages — each is dropped with a warning and the rest of the batch still decodes

Examples

yaml

input:
  type: kafka
  brokers:
    - localhost:9092
  topics:
    - users
  consumer_group: arkflow
  start_from_latest: false
  codec:
    type: protobuf
    message_type: message.Message
    proto_inputs:
      - ./examples/message.proto

yaml

codec:
  type: protobuf
  message_type: message.Message
  proto_inputs:
    - ./examples/message.proto

Notes

• Only proto3 scalar fields are supported: bool, int32/sint32/sfixed32, int64/sint64/sfixed64, uint32/fixed32, uint64/fixed64, float, double, string, bytes, and enum (mapped to Arrow Int32).
• Nested messages, repeated, map, oneof, and proto3 optional fields are not supported; encountering them returns an error during encoding/decoding.
• When decoding multiple messages, the batch is merged (concat_batches) using the schema of the first message; fields across messages must be compatible.