0

I am running Kafka/Zookeeper on my Mac; Kafka works fine: I can create topics and send/receive messages to them using the console consumer. However, when trying to start KSQL from a Docker container it does not connect to Kafka

Here are the zookeeper and kafka properties:

Mac: ~/KAFKA $ grep -v ^# /usr/local/homebrew/etc/kafka/zookeeper.properties
dataDir=/usr/local/homebrew/var/lib/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false
Mac: ~/KAFKA $ 
Mac: ~/KAFKA $ grep -v ^# /usr/local/homebrew/etc/kafka/server.properties
broker.id=0
listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/usr/local/homebrew/var/lib/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
Mac: ~/KAFKA $

This is the docker-compose.yml file:

Mac: ~/KSQL $ cat docker-compose.yml 
---
version: '2'

services:
  ksqldb-server:
    image: confluentinc/ksqldb-server:0.15.0
    hostname: ksqldb-server
    container_name: ksqldb-server
    ports:
      - "8088:8088"
      - "9092:9092"
    environment:
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_BOOTSTRAP_SERVERS: $BROKER_ENDPOINT
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"

  ksqldb-cli:
    image: confluentinc/ksqldb-cli:0.15.0
    container_name: ksqldb-cli
    depends_on:
      - ksqldb-server
    entrypoint: /bin/sh
    tty: true
Mac: ~/KSQL $

The KSQL container starts but does not connect to Kafka:

Mac: ~/KSQL $ docker-compose up                                                                
Recreating ksqldb-server ... done
Recreating ksqldb-cli    ... done
Attaching to ksqldb-server, ksqldb-cli
ksqldb-server    | ===> Configuring ksqlDB...
ksqldb-server    | ===> Launching ksqlDB Server...
ksqldb-server    | OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
ksqldb-server    | [2021-04-08 14:29:09,560] INFO KsqlConfig values: 
ksqldb-server    |  ksql.access.validator.enable = auto
ksqldb-server    |  ksql.authorization.cache.expiry.time.secs = 30
ksqldb-server    |  ksql.authorization.cache.max.entries = 10000
ksqldb-server    |  ksql.cast.strings.preserve.nulls = true
ksqldb-server    |  ksql.connect.url = http://localhost:8083
ksqldb-server    |  ksql.connect.worker.config = 
ksqldb-server    |  ksql.create.or.replace.enabled = true
ksqldb-server    |  ksql.error.classifier.regex = 
ksqldb-server    |  ksql.extension.dir = ext
ksqldb-server    |  ksql.hidden.topics = [_confluent.*, __confluent.*, _schemas, __consumer_offsets, __transaction_state, connect-configs, connect-offsets, connect-status, connect-statuses]
ksqldb-server    |  ksql.insert.into.values.enabled = true
ksqldb-server    |  ksql.internal.topic.min.insync.replicas = 1
ksqldb-server    |  ksql.internal.topic.replicas = 1
ksqldb-server    |  ksql.metastore.backup.location = 
ksqldb-server    |  ksql.metrics.extension = null
ksqldb-server    |  ksql.metrics.tags.custom = 
ksqldb-server    |  ksql.output.topic.name.prefix = 
ksqldb-server    |  ksql.persistence.default.format.key = KAFKA
ksqldb-server    |  ksql.persistence.default.format.value = null
ksqldb-server    |  ksql.persistence.wrap.single.values = null
ksqldb-server    |  ksql.persistent.prefix = query_
ksqldb-server    |  ksql.properties.overrides.denylist = []
ksqldb-server    |  ksql.pull.queries.enable = true
ksqldb-server    |  ksql.query.error.max.queue.size = 10
ksqldb-server    |  ksql.query.persistent.active.limit = 2147483647
ksqldb-server    |  ksql.query.persistent.max.bytes.buffering.total = -1
ksqldb-server    |  ksql.query.pull.enable.standby.reads = false
ksqldb-server    |  ksql.query.pull.max.allowed.offset.lag = 9223372036854775807
ksqldb-server    |  ksql.query.pull.max.qps = 2147483647
ksqldb-server    |  ksql.query.pull.metrics.enabled = false
ksqldb-server    |  ksql.query.pull.thread.pool.size = 100
ksqldb-server    |  ksql.query.retry.backoff.initial.ms = 15000
ksqldb-server    |  ksql.query.retry.backoff.max.ms = 900000
ksqldb-server    |  ksql.query.status.running.threshold.seconds = 300
ksqldb-server    |  ksql.query.transient.max.bytes.buffering.total = -1
ksqldb-server    |  ksql.readonly.topics = [_confluent.*, __confluent.*, _schemas, __consumer_offsets, __transaction_state, connect-configs, connect-offsets, connect-status, connect-statuses]
ksqldb-server    |  ksql.schema.registry.url = 
ksqldb-server    |  ksql.security.extension.class = null
ksqldb-server    |  ksql.service.id = default_
ksqldb-server    |  ksql.sink.window.change.log.additional.retention = 1000000
ksqldb-server    |  ksql.streams.shutdown.timeout.ms = 300000
ksqldb-server    |  ksql.suppress.buffer.size.bytes = -1
ksqldb-server    |  ksql.suppress.enabled = false
ksqldb-server    |  ksql.timestamp.throw.on.invalid = false
ksqldb-server    |  ksql.transient.prefix = transient_
ksqldb-server    |  ksql.udf.collect.metrics = false
ksqldb-server    |  ksql.udf.enable.security.manager = true
ksqldb-server    |  ksql.udfs.enabled = true
ksqldb-server    |  ksql.variable.substitution.enable = true
ksqldb-server    |  metric.reporters = []
ksqldb-server    |  ssl.cipher.suites = null
ksqldb-server    |  ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ksqldb-server    |  ssl.endpoint.identification.algorithm = https
ksqldb-server    |  ssl.engine.factory.class = null
ksqldb-server    |  ssl.key.password = null
ksqldb-server    |  ssl.keymanager.algorithm = SunX509
ksqldb-server    |  ssl.keystore.certificate.chain = null
ksqldb-server    |  ssl.keystore.key = null
ksqldb-server    |  ssl.keystore.location = null
ksqldb-server    |  ssl.keystore.password = null
ksqldb-server    |  ssl.keystore.type = JKS
ksqldb-server    |  ssl.protocol = TLSv1.3
ksqldb-server    |  ssl.provider = null
ksqldb-server    |  ssl.secure.random.implementation = null
ksqldb-server    |  ssl.trustmanager.algorithm = PKIX
ksqldb-server    |  ssl.truststore.certificates = null
ksqldb-server    |  ssl.truststore.location = null
ksqldb-server    |  ssl.truststore.password = null
ksqldb-server    |  ssl.truststore.type = JKS
ksqldb-server    |  (io.confluent.ksql.util.KsqlConfig:372)
ksqldb-server    | [2021-04-08 14:29:09,601] INFO KsqlRestConfig values: 
ksqldb-server    |  access.control.allow.headers = []
ksqldb-server    |  access.control.allow.methods = []
ksqldb-server    |  access.control.allow.origin = 
ksqldb-server    |  authentication.method = NONE
ksqldb-server    |  authentication.realm = 
ksqldb-server    |  authentication.roles = [*]
ksqldb-server    |  authentication.skip.paths = []
ksqldb-server    |  ksql.advertised.listener = null
ksqldb-server    |  ksql.authentication.plugin.class = null
ksqldb-server    |  ksql.healthcheck.interval.ms = 5000
ksqldb-server    |  ksql.heartbeat.check.interval.ms = 200
ksqldb-server    |  ksql.heartbeat.discover.interval.ms = 2000
ksqldb-server    |  ksql.heartbeat.enable = false
ksqldb-server    |  ksql.heartbeat.missed.threshold.ms = 3
ksqldb-server    |  ksql.heartbeat.send.interval.ms = 100
ksqldb-server    |  ksql.heartbeat.thread.pool.size = 3
ksqldb-server    |  ksql.heartbeat.window.ms = 2000
ksqldb-server    |  ksql.internal.listener = null
ksqldb-server    |  ksql.internal.ssl.client.authentication = NONE
ksqldb-server    |  ksql.lag.reporting.enable = false
ksqldb-server    |  ksql.lag.reporting.send.interval.ms = 5000
ksqldb-server    |  ksql.local.commands.location = 
ksqldb-server    |  ksql.logging.server.rate.limited.request.paths = 
ksqldb-server    |  ksql.logging.server.rate.limited.response.codes = 
ksqldb-server    |  ksql.max.push.queries = 100
ksqldb-server    |  ksql.server.command.blocked.threshold.error.ms = 15000
ksqldb-server    |  ksql.server.command.response.timeout.ms = 5000
ksqldb-server    |  ksql.server.error.messages = class io.confluent.ksql.rest.DefaultErrorMessages
ksqldb-server    |  ksql.server.exception.uncaught.handler.enable = false
ksqldb-server    |  ksql.server.install.dir = /usr
ksqldb-server    |  ksql.server.preconditions = []
ksqldb-server    |  ksql.server.websockets.num.threads = 5
ksqldb-server    |  ksql.ssl.keystore.alias.external = 
ksqldb-server    |  ksql.ssl.keystore.alias.internal = 
ksqldb-server    |  ksql.verticle.instances = 16
ksqldb-server    |  ksql.worker.pool.size = 100
ksqldb-server    |  listeners = [http://0.0.0.0:8088]
ksqldb-server    |  query.stream.disconnect.check = 1000
ksqldb-server    |  ssl.cipher.suites = []
ksqldb-server    |  ssl.client.auth = false
ksqldb-server    |  ssl.client.authentication = NONE
ksqldb-server    |  ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ksqldb-server    |  ssl.keystore.location = 
ksqldb-server    |  ssl.keystore.password = [hidden]
ksqldb-server    |  ssl.keystore.reload = false
ksqldb-server    |  ssl.keystore.type = JKS
ksqldb-server    |  ssl.keystore.watch.location = 
ksqldb-server    |  ssl.truststore.location = 
ksqldb-server    |  ssl.truststore.password = [hidden]
ksqldb-server    |  ssl.truststore.type = JKS
ksqldb-server    |  (io.confluent.ksql.rest.server.KsqlRestConfig:372)
ksqldb-server    | [2021-04-08 14:29:09,612] INFO KsqlConfig values: 
ksqldb-server    |  ksql.access.validator.enable = auto
ksqldb-server    |  ksql.authorization.cache.expiry.time.secs = 30
ksqldb-server    |  ksql.authorization.cache.max.entries = 10000
ksqldb-server    |  ksql.cast.strings.preserve.nulls = true
ksqldb-server    |  ksql.connect.url = http://localhost:8083
ksqldb-server    |  ksql.connect.worker.config = 
ksqldb-server    |  ksql.create.or.replace.enabled = true
ksqldb-server    |  ksql.error.classifier.regex = 
ksqldb-server    |  ksql.extension.dir = ext
ksqldb-server    |  ksql.hidden.topics = [_confluent.*, __confluent.*, _schemas, __consumer_offsets, __transaction_state, connect-configs, connect-offsets, connect-status, connect-statuses]
ksqldb-server    |  ksql.insert.into.values.enabled = true
ksqldb-server    |  ksql.internal.topic.min.insync.replicas = 1
ksqldb-server    |  ksql.internal.topic.replicas = 1
ksqldb-server    |  ksql.metastore.backup.location = 
ksqldb-server    |  ksql.metrics.extension = null
ksqldb-server    |  ksql.metrics.tags.custom = 
ksqldb-server    |  ksql.output.topic.name.prefix = 
ksqldb-server    |  ksql.persistence.default.format.key = KAFKA
ksqldb-server    |  ksql.persistence.default.format.value = null
ksqldb-server    |  ksql.persistence.wrap.single.values = null
ksqldb-server    |  ksql.persistent.prefix = query_
ksqldb-server    |  ksql.properties.overrides.denylist = []
ksqldb-server    |  ksql.pull.queries.enable = true
ksqldb-server    |  ksql.query.error.max.queue.size = 10
ksqldb-server    |  ksql.query.persistent.active.limit = 2147483647
ksqldb-server    |  ksql.query.persistent.max.bytes.buffering.total = -1
ksqldb-server    |  ksql.query.pull.enable.standby.reads = false
ksqldb-server    |  ksql.query.pull.max.allowed.offset.lag = 9223372036854775807
ksqldb-server    |  ksql.query.pull.max.qps = 2147483647
ksqldb-server    |  ksql.query.pull.metrics.enabled = false
ksqldb-server    |  ksql.query.pull.thread.pool.size = 100
ksqldb-server    |  ksql.query.retry.backoff.initial.ms = 15000
ksqldb-server    |  ksql.query.retry.backoff.max.ms = 900000
ksqldb-server    |  ksql.query.status.running.threshold.seconds = 300
ksqldb-server    |  ksql.query.transient.max.bytes.buffering.total = -1
ksqldb-server    |  ksql.readonly.topics = [_confluent.*, __confluent.*, _schemas, __consumer_offsets, __transaction_state, connect-configs, connect-offsets, connect-status, connect-statuses]
ksqldb-server    |  ksql.schema.registry.url = 
ksqldb-server    |  ksql.security.extension.class = null
ksqldb-server    |  ksql.service.id = default_
ksqldb-server    |  ksql.sink.window.change.log.additional.retention = 1000000
ksqldb-server    |  ksql.streams.shutdown.timeout.ms = 300000
ksqldb-server    |  ksql.suppress.buffer.size.bytes = -1
ksqldb-server    |  ksql.suppress.enabled = false
ksqldb-server    |  ksql.timestamp.throw.on.invalid = false
ksqldb-server    |  ksql.transient.prefix = transient_
ksqldb-server    |  ksql.udf.collect.metrics = false
ksqldb-server    |  ksql.udf.enable.security.manager = true
ksqldb-server    |  ksql.udfs.enabled = true
ksqldb-server    |  ksql.variable.substitution.enable = true
ksqldb-server    |  metric.reporters = []
ksqldb-server    |  ssl.cipher.suites = null
ksqldb-server    |  ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ksqldb-server    |  ssl.endpoint.identification.algorithm = https
ksqldb-server    |  ssl.engine.factory.class = null
ksqldb-server    |  ssl.key.password = null
ksqldb-server    |  ssl.keymanager.algorithm = SunX509
ksqldb-server    |  ssl.keystore.certificate.chain = null
ksqldb-server    |  ssl.keystore.key = null
ksqldb-server    |  ssl.keystore.location = null
ksqldb-server    |  ssl.keystore.password = null
ksqldb-server    |  ssl.keystore.type = JKS
ksqldb-server    |  ssl.protocol = TLSv1.3
ksqldb-server    |  ssl.provider = null
ksqldb-server    |  ssl.secure.random.implementation = null
ksqldb-server    |  ssl.trustmanager.algorithm = PKIX
ksqldb-server    |  ssl.truststore.certificates = null
ksqldb-server    |  ssl.truststore.location = null
ksqldb-server    |  ssl.truststore.password = null
ksqldb-server    |  ssl.truststore.type = JKS
ksqldb-server    |  (io.confluent.ksql.util.KsqlConfig:372)
ksqldb-server    | [2021-04-08 14:29:10,329] WARN The configuration 'metrics.context.resource.version' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:380)
ksqldb-server    | [2021-04-08 14:29:10,329] WARN The configuration 'metrics.context.resource.version' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:380)
ksqldb-server    | [2021-04-08 14:29:10,330] WARN The configuration 'metrics.context.resource.commit.id' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:380)
ksqldb-server    | [2021-04-08 14:29:10,330] WARN The configuration 'metrics.context.resource.commit.id' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:380)
ksqldb-server    | [2021-04-08 14:29:10,359] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
ksqldb-server    | [2021-04-08 14:29:10,359] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
ksqldb-server    | [2021-04-08 14:29:10,466] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
ksqldb-server    | [2021-04-08 14:29:10,466] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
ksqldb-server    | [2021-04-08 14:29:10,671] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
ksqldb-server    | [2021-04-08 14:29:10,671] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:773)
... etc.

What am I doing wrong ???

Alex V
  • 1

1 Answers1

0

KSQL is trying to connect to 127.0.0.1:9092. It should be connecting to the host instead. Setting KSQL_BOOTSTRAP_SERVERS to host machine's IP/port should fix this. Set bootstrap servers as shown below:

KSQL_BOOTSTRAP_SERVERS: ${host.docker.internal}:9092
mrnakumar
  • 625
  • 6
  • 13
  • 1
    It still does not work: Mac: ~/KSQL $ cat docker-compose.yml --- ... environment: KSQL_LISTENERS: http://0.0.0.0:8088 KSQL_BOOTSTRAP_SERVERS: ${host.docker.internal}:9092 ... Mac: ~/KSQL $ docker-compose up ERROR: Invalid interpolation format for "environment" option in service "ksqldb-server-sa": "${host.docker.internal}:9092" Mac: ~/KSQL $ Where can I specify host.docker.internal ?? – Alex V Apr 09 '21 at 20:31
  • Can you try changing version to '3' in file 'docker-compose.yml' – mrnakumar Apr 11 '21 at 22:00
  • 1
    it does not help I need to tell docker to map its internal ports to the host ports. The 'ports' section in docker-compose.yml does not appear to be sufficient: --- version: '3' services: ksqldb-server-sa: image: confluentinc/ksqldb-server hostname: ksqldb-server-sa container_name: ksqldb-server-sa ports: - "8088:8088" - "9092:9092" environment: KSQL_LISTENERS: http://0.0.0.0:8088 KSQL_BOOTSTRAP_SERVERS: $BROKER_ENDPOINT ... or else I need to specify BROKER_ENDOINT differently: $ cat .env export BROKER_ENDPOINT=localhost:9092 – Alex V Apr 13 '21 at 18:14
  • This wont work because Kafka itself is telling clients to connect on localhost, not on `host.docker.internal`. See duplicate post – OneCricketeer Jul 11 '23 at 15:22