Problems setting up KoP with Pulsar 3.3.4

[paulo-alves-flttr]

I have created a custom Pulsar image from Pulsar 3.3.4 that adds pulsar-protocol-handler-kafka-3.1.1.1.nar file to /pulsar/protocols/ and adds the following lines to .conf files:

kafkaListeners=PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
kafkaAdvertisedListeners=PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
kafkaNamespace=kafka
kafkaTransactionCoordinatorEnabled=true
kopSchemaRegistryEnable=true
kopSchemaRegistryPort=8081
messagingProtocols=kafka
protocolHandlerDirectory=./protocols

A have a docker compose like the following:

networks:
  pulsar:
    driver: bridge
services:
  # Start zookeeper
  zookeeper:
    image: taulin/pulsar-with-kop:0.0.9
    container_name: zookeeper
    restart: on-failure
    networks:
      - pulsar
    volumes:
      - ./data/zookeeper:/pulsar/data/zookeeper
    environment:
      - metadataStoreUrl=zk:zookeeper:2181
      - PULSAR_MEM=-Xms256m -Xmx256m -XX:MaxDirectMemorySize=256m
    command: |
      bash -c "bin/apply-config-from-env.py conf/zookeeper.conf && \
             bin/generate-zookeeper-config.sh conf/zookeeper.conf && \
             exec bin/pulsar zookeeper"
    healthcheck:
      test: ["CMD", "bin/pulsar-zookeeper-ruok.sh"]
      interval: 10s
      timeout: 5s
      retries: 30

  # Init cluster metadata
  pulsar-init:
    container_name: pulsar-init
    hostname: pulsar-init
    image: taulin/pulsar-with-kop:0.0.9
    networks:
      - pulsar
    command: bash -c "bin/pulsar initialize-cluster-metadata --cluster cluster-a --zookeeper zookeeper:2181 --configuration-store zookeeper:2181 --web-service-url http://broker:8080 --broker-service-url pulsar://broker:6650"
    depends_on:
      zookeeper:
        condition: service_healthy

  # Start bookie
  bookie:
    image: taulin/pulsar-with-kop:0.0.9
    container_name: bookie
    restart: on-failure
    networks:
      - pulsar
    environment:
      - clusterName=cluster-a
      - zkServers=zookeeper:2181
      - metadataServiceUri=metadata-store:zk:zookeeper:2181
      # otherwise every time we run docker compose uo or down we fail to start due to Cookie
      # See: https://github.com/apache/bookkeeper/blob/405e72acf42bb1104296447ea8840d805094c787/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java#L57-68
      - advertisedAddress=bookie
      - BOOKIE_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m
    depends_on:
      zookeeper:
        condition: service_healthy
      pulsar-init:
        condition: service_completed_successfully
    # Map the local directory to the container to avoid bookie startup failure due to insufficient container disks.
    volumes:
      - ./data/bookkeeper:/pulsar/data/bookkeeper
    command: bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie"

  # Start broker
  broker:
    image: taulin/pulsar-with-kop:0.0.9
    container_name: broker
    hostname: broker
    restart: on-failure
    networks:
      - pulsar
    environment:
      - metadataStoreUrl=zk:zookeeper:2181
      - zookeeperServers=zookeeper:2181
      - clusterName=cluster-a
      - managedLedgerDefaultEnsembleSize=1
      - managedLedgerDefaultWriteQuorum=1
      - managedLedgerDefaultAckQuorum=1
      - advertisedAddress=broker
      - advertisedListeners=external:pulsar://127.0.0.1:6650
      - PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m
    depends_on:
      zookeeper:
        condition: service_healthy
      bookie:
        condition: service_started
    ports:
      - "6650:6650"
      - "8080:8080"
      - "9092:9092"
      - "8081:8081"
    expose:
      - '29092'
      - '28081'
    command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker"

I am able to docker compose up with success but when I try producing Kafka messages through kafka-console-producer with:

kafka-console-producer --broker-list localhost:9092 --topic test

I get the following output error:

[2025-05-09 14:17:33,466] WARN [Producer clientId=console-producer] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

Can someone help me spot what I did wrong?