search

twmb / franz-go

franz-go contains a feature complete, pure Go library for interacting with Kafka from 0.8.0 through 3.8+. Producing, consuming, transacting, administrating, etc.
BSD 3-Clause "New" or "Revised" License
1.8k stars 183 forks source link

fetchMetadata causing issues when running kafka in docker during dev #611

Closed ghstahl closed 11 months ago

ghstahl commented 11 months ago

I am running kafka in docker.
Inside docker its kafka:9092 Outside of docker kafka:9092 means nothing.

I am using benthos's kafka_franz input plugin.
I am debugging this app using visual studio code so my app's URL to Kafka is localhost:9092 I have a kafka_franz producer in a separate app that works just fine when I give it the localhost:9092 seed.

input:
  kafka_franz:
    seed_brokers: ["localhost:9092"]
    topics: ["cloudevents-core"]
    regexp_topics: false
    consumer_group: "$Default"
    batching:
      count: 2
      period: 1m
      processors:
        - archive:
            format: json_array

The benthos implementation is a consumer. Somewhere in the chain a call to fetchMetadata results in pulling kafka.9092 from the server.

Why is this happening at all? I would want my seeds to be the final authority.

This results in warnings from the library.

level=warning msg="unable to open connection to broker" @service=benthos addr="kafka:9092" broker=1 err="dial tcp: lookup kafka: no such host" label="" path=root.input

You can repo this just by using benthos as is with the above config.

twmb commented 11 months ago

Seeds are only used for the initial discovery, the metadata response returns all broker-configured advertised listeners -- and the advertised listeners are used thereafter by the client. The franz-go client does rotate through seed brokers as well occasionally, but the expectation per the Kafka protocol is to use discovered brokers from metadata responses.

There are more reasons for this -- what node ID is a seed broker? What happens when you scale your cluster up or down -- do you need to change seed brokers? etc.

ghstahl commented 11 months ago

How do you develop with just having a simple kafka docker-compose? Is there some config you use to tell kakfa it's really localhost:9092 when a call is made to it?

twmb commented 11 months ago

I do something like this, where kafka and zk are my own images (not much to them):

version: "3.7"
services:
  zookeeper:
    image: twmb:zk-3.9.1
    network_mode: host

  kafka-1:
    image: twmb:kafka-2.13-3.6.0
    depends_on:
      - zookeeper
    network_mode: host
    environment:
      KPROP_BROKER_ID: 1
      KPROP_LISTENERS: 'PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093'
      KPROP_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
      KPROP_DELEGATION_TOKEN_MASTER_KEY: 'super_secret'
      KPROP_SUPER_USERS: 'User:ANONYMOUS;User:adminscram;User:adminplain'
      KPROP_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
      KAFKA_OPTS: "-Djava.security.auth.login.config=/kafka/config/kafka_server_jaas.conf"
      KPROP_TRANSACTION_PARTITION_VERIFICATION_ENABLE: false
    volumes:
      - type: bind
        source: ${PWD}/kcfg/kafka_server_jaas.conf
        target: /kafka/config/kafka_server_jaas.conf
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/log4j.properties
        target: /kafka/config/log4j.properties
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/setup.sh
        target: /kafka/setup.sh
        read_only: true

  kafka-2:
    image: twmb:kafka-2.13-3.6.0
    depends_on:
      - zookeeper
    network_mode: host
    environment:
      KPROP_BROKER_ID: 2
      KPROP_LISTENERS: 'PLAINTEXT://localhost:9094,SASL_PLAINTEXT://localhost:9095'
      KPROP_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
      KPROP_DELEGATION_TOKEN_MASTER_KEY: 'super_secret'
      KPROP_SUPER_USERS: 'User:ANONYMOUS;User:adminscram;User:adminplain'
      KPROP_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
      KAFKA_OPTS: "-Djava.security.auth.login.config=/kafka/config/kafka_server_jaas.conf"
      KPROP_TRANSACTION_PARTITION_VERIFICATION_ENABLE: false
    volumes:
      - type: bind
        source: ${PWD}/kcfg/kafka_server_jaas.conf
        target: /kafka/config/kafka_server_jaas.conf
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/log4j.properties
        target: /kafka/config/log4j.properties
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/setup.sh
        target: /kafka/setup.sh
        read_only: true

  kafka-3:
    image: twmb:kafka-2.13-3.6.0
    depends_on:
      - zookeeper
    network_mode: host
    environment:
      KPROP_BROKER_ID: 3
      KPROP_LISTENERS: 'PLAINTEXT://localhost:9096,SASL_PLAINTEXT://localhost:9097'
      KPROP_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
      KPROP_DELEGATION_TOKEN_MASTER_KEY: 'super_secret'
      KPROP_SUPER_USERS: 'User:ANONYMOUS;User:adminscram;User:adminplain'
      KPROP_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
      KAFKA_OPTS: "-Djava.security.auth.login.config=/kafka/config/kafka_server_jaas.conf"
      KPROP_TRANSACTION_PARTITION_VERIFICATION_ENABLE: false
    volumes:
      - type: bind
        source: ${PWD}/kcfg/kafka_server_jaas.conf
        target: /kafka/config/kafka_server_jaas.conf
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/log4j.properties
        target: /kafka/config/log4j.properties
        read_only: true
      - type: bind
        source: ${PWD}/kcfg/setup.sh
        target: /kafka/setup.sh
        read_only: true

I also occasionally test with bitnami:

version: "3.7"
services:
  kafka:
    image: bitnami/kafka:latest
    network_mode: host
    environment:
      KAFKA_ENABLE_KRAFT: yes
      KAFKA_CFG_PROCESS_ROLES: controller,broker
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@127.0.0.1:9093
      KAFKA_CFG_TRANSACTION_PARTITION_VERIFICATION_ENABLE: false
      # Set this to "PLAINTEXT://127.0.0.1:9092" if you want to run this container on localhost via Docker
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CFG_NODE_ID: 1
      ALLOW_PLAINTEXT_LISTENER: yes
      KAFKA_KRAFT_CLUSTER_ID: XkpGZQ27R3eTl3OdTm2LYA # 16 byte base64-encoded UUID

I don't do this on Mac because I've never really cared to figure out networking on Mac

ghstahl commented 11 months ago

DUH! I was dipping into the Halloween candy and coming down off of a sugar coma.

Thanks for posting your yaml.

I was using the wrong port. My advertised listener EXTERNAL://localhost:9093

 kafka:
    <<: *default-service
    image: "bitnami/kafka:3.3.2-debian-11-r186"
    container_name: "${KAFKA_NAME}"
    ports:
      - ${KAFKA_PORT}:9092 # Internal to the mapped-in-a-box network
      - "9093:9093" # External to the host machine (host.docker.internal)
      - "0.0.0.0:9094:9094" # External to the host machine (host.docker.internal)
    environment:
      - KAFKA_BROKER_ID=1
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT,ONION:PLAINTEXT
      - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093,ONION://:9094
      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://${KAFKA_NAME}:9092,EXTERNAL://localhost:9093,ONION://kafkaonion:9094
      - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
      - KAFKA_ENABLE_KRAFT=no
      # TODO: newer containers have permission problems with those mounts
      #volumes:
      #- ./config/kafka/log4j.properties:/opt/bitnami/kafka/config/log4j.properties:rw
      #- ./config/kafka/server.properties:/opt/bitnami/kafka/config/server.properties:rw
    depends_on:
      - zookeeper
    healthcheck:
      test: [ "CMD-SHELL", "kafka-broker-api-versions.sh --version" ]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s