Add support to length delimited protobuf messages

[meirdev]

Hi,

I tried to read protobuf messages with a length-delimited prefix, and I got the following error:

To fix the problem I think it is necessary to add a new field to the schema configuration to determine whether the producer sends messages with a length-delimited prefix or not:

---

My topic listens to messages sent by GoFlow, this is my docker-compose.yml (I added nflow-generator for the demo):

services:
  kafka:
    image: bitnami/kafka:4.0.0
    restart: always
    ports:
      - "9092:9092"
    environment:
      - KAFKA_DELETE_TOPIC_ENABLE=true
      - KAFKA_KRAFT_CLUSTER_ID=AAAAAAAAAAAAAAAAAAAAAA
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_ENABLE_KRAFT=yes
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_CFG_BROKER_ID=1
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_LISTENERS=CONTROLLER://localhost:9091,HOST://0.0.0.0:9092,INTERNAL://0.0.0.0:9093
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,HOST:PLAINTEXT
      - KAFKA_CFG_ADVERTISED_LISTENERS=HOST://127.0.0.1:9092,INTERNAL://kafka:9093
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@localhost:9091

  goflow2:
    image: netsampler/goflow2:latest
    ports:
      - "6343:6343/udp"
      - "2055:2055/udp"
      - "8080:8080"
    restart: always
    command:
      - -transport.kafka.brokers=kafka:9093
      - -transport=kafka
      - -transport.kafka.topic=flows
      - -format=bin
    depends_on:
      - kafka

  nflow-generator:
    image: networkstatic/nflow-generator
    restart: always
    command: -p 2055 -t goflow2
    depends_on:
      - goflow2

  arroyo:
    image: ghcr.io/arroyosystems/arroyo:latest
    restart: always
    ports:
      - "5115:5115"
    depends_on:
      - kafka

[mwylde]

Thanks, this is great!