Support separation of Internal and External traffic by Broker (KIP-103)

[PHameete]

Currently, kafka-node does not have support for brokers that are configured to have separate listeners for internal and external traffic. This was introduced in KIP-103.

There are 2 aspects to this KIP that are important:

1. The protocol of the url's in the endpoints listed in the broker metadata is not necessarily the used protocol. It can instead be the name of the listener which must then be mapped to the protocol via the 'listener_security_protocol_map' metadata that was newly introduced in KIP-103.
2. The kafka-node adapter currently selects the first endpoint that supports the protocol that is selected for usage by the kafka-node adapter. However, there may be multiple endpoints that use this protocol: for example both an INTERNAL as well as an EXTERNAL endpoint. In this case the EXTERNAL endpoint should be chosen as the endpoint that is used because the INTERNAL endpoint can not be reached by the client.

Reproduction

To reproduce this, set up a Kafka broker that separates INTERNAL and EXTERNAL traffic. You can do this by setting up a Kafka Broker using Docker and using the following configuration:

---
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper
    hostname: zookeeper
    ports:
      - "3500:3500"
    environment:
      ZOOKEEPER_CLIENT_PORT: 3500
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka
    hostname: broker
    depends_on:
      - zookeeper
    ports:
      - "3501:3501"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:3500'
      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://broker:9092,EXTERNAL://localhost:3501'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,EXTERNAL://0.0.0.0:3501'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      KAFKA_MESSAGE_MAX_BYTES: 100000000
      KAFKA_REPLICA_FETCH_MAX_BYTES: 100000000

The endpoint broker:9092 is ment to be used used internally between the containers. The endpoint localhost:3501 is used by external Kafka clients to connect to the broker.

If you now run the example consumer.js for zookeeper:3500 no suitable endpoint can be found because the advertised listeners with names EXTERNAL and INTERNAL are not mapped to the protocol PLAINTEXT. Moreover, if this mapping were in place, it would select the INTERNAL endpoint as a first match (see client.js line 126, function setupBrokerProfiles). However, the INTERNAL endpoint can not be used to access the broker from outside the Docker bridge network.

[hyperlink]

Did merging your PR resolve this issue?

[PHameete]

@hyperlink yes, but only my first point. Not point 2:

The kafka-node adapter currently selects the first endpoint that supports the protocol that is selected for usage by the kafka-node adapter. However, there may be multiple endpoints that use this protocol: for example both an INTERNAL as well as an EXTERNAL endpoint. In this case the EXTERNAL endpoint should be chosen as the endpoint that is used because the INTERNAL endpoint can not be reached by the client.

So only if your ADVERTISED_LISTENERS states the endpoint to be used by kafka-node first. To resolve this some sort of mechanism must be built that selects the first endpoint that actually works. Because a kafka broker can have multiple endpoints, not all of which will be reachable necessarily.

[stevesirois]

Is there a will to solve this issue in a near future? I confirm the problem in this same scenario (Docker env.)

[kortatu]

I know that ZkOptions are meant to be passed directly to Zookeeper, but as this listener selection problem only happens when using Zookeeper, maybe a new preferredListener option could be added to ZkOptions so if it is provided by the developer, that endpoint would be selected (EXTERNAL or OUTSIDE, or whatever name the listener was given) instead of the first one that matches the protocol.

[kortatu]

I have created a workaround branching from version 3.0.1 and using ZkOptions as I said before to selecte prederred listeners:
https://github.com/epiclabs-io/kafka-node/tree/v3.0.1_preferred_listener