unable to write on kafka topic created on kafka container

Question

I am writing integration tests and created kafka topic using docker command docker exec kafka-broker kafka-topics.sh --create --bootstrap-server localhost:9093 --partitions 1 --replication-factor 1 --topic test-topic in github workflow, Topic is created successfully. Using kafka topic in testcase to write data on but it gives error org.apache.kafka.common.KafkaException: Failed to construct kafka producer

Full erro message

WARN ClientUtils: Couldn't resolve server kafka:9092 from bootstrap.servers as DNS resolution failed for kafka org.apache.kafka.common.KafkaException: Failed to construct kafka producer at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:440) at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:291) at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:274) at org.apache.spark.sql.kafka010.producer.InternalKafkaProducerPool.createKafkaProducer(InternalKafkaProducerPool.scala:136) at org.apache.spark.sql.kafka010.producer.InternalKafkaProducerPool.$anonfun$acquire$1(InternalKafkaProducerPool.scala:83) at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86) at org.apache.spark.sql.kafka010.producer.InternalKafkaProducerPool.acquire(InternalKafkaProducerPool.scala:82) at org.apache.spark.sql.kafka010.producer.InternalKafkaProducerPool$.acquire(InternalKafkaProducerPool.scala:198) at org.apache.spark.sql.kafka010.KafkaWriteTask.execute(KafkaWriteTask.scala:49) at org.apache.spark.sql.kafka010.KafkaWriter$.$anonfun$write$2(KafkaWriter.scala:72) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504) at org.apache.spark.sql.kafka010.KafkaWriter$.$anonfun$write$1(KafkaWriter.scala:73) at org.apache.spark.sql.kafka010.KafkaWriter$.$anonfun$write$1$adapted(KafkaWriter.scala:70) at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:1011) at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:1011) at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2278) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at org.apache.spark.scheduler.Task.run(Task.scala:136) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:89) at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48) at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:414)

I have a code as bellow

df = #df creation code
    df.write.format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("kafka.security.protocol", "PLAINTEXT")
        .option("topic", test-topic)
        .save()

i want to run testcases in CICD. Can someone please help me how how to fix this?

Answer 1

DNS resolution failed for kafka ... No resolvable bootstrap urls given

Outside of the Docker network, you cannot resolve container names.

Option 1: you need to run tests outside of a container and use localhost:9093, the external listener

Option 2: Run your tests in a container, on a shared Docker network (using --network flag for Docker run, or using Docker Compose) and use kafka:9092, where the host address can be resolved, the internal listener

in github workflow

Those always run as containers, so you're limited to option 2. --network flag isn't available there, however.

Start here https://testcontainers-python.readthedocs.io/en/latest/kafka/README.html

And also see https://docs.github.com/en/actions/using-containerized-services/about-service-containers#communicating-with-service-containers

Answer 2

Looks like there is a port number mismatch: try changing .option("kafka.bootstrap.servers", kafka:9092) to .option("kafka.bootstrap.servers", "localhost:9093") - from the topic creation command it looks like your bootstrap server is setup to run on port 9093 & not 9092.