I want to make a Kafka producer send messages to a topic in a Kafka broker, runned in Docker, using org.apache.kafka Java library. Here’s what I’ve done:
I started a Kafka and a Zookeeper server using docker-compose up -d with this docker-compose.yml file:
version: '2'
services:
zookeeper:
image: 'bitnami/zookeeper:3'
ports:
- '2181:2181'
volumes:
- 'zookeeper_data:/bitnami'
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
kafka:
image: 'bitnami/kafka:2'
ports:
- '9092:9092'
- '29092:29092'
volumes:
- 'kafka_data:/bitnami'
environment:
- KAFKA_BROKER_ID=1
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:29092
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
- KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
- ALLOW_PLAINTEXT_LISTENER=yes
depends_on:
- zookeeper
volumes:
zookeeper_data:
driver: local
kafka_data:
driver: local
Output of docker-compose ps:
Name Command State Ports
-----------------------------------------------------------------------------------------------------------------
dockertoolbox_kafka_1 /entrypoint.sh /run.sh Up 0.0.0.0:29092->29092/tcp, 0.0.0.0:9092->9092/tcp
dockertoolbox_zookeeper_1 /entrypoint.sh /run.sh Up 0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp, 8080/tcp
Then, I successfully created a topic via Docker terminal:
docker exec -it dockertoolbox_kafka_1 kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic testTopic
The problem is when I try to connect a KafkaProducer from an external Java app (code taken from https://www.tutorialkart.com/apache-kafka/producer-example-in-apache-kafka/ and edited the port):
public class SampleProducer extends Thread {
private final KafkaProducer<Integer, String> producer;
private final String topic;
private final Boolean isAsync;
public static final String KAFKA_SERVER_URL = "localhost";
public static final int KAFKA_SERVER_PORT = 29092;
public static final String CLIENT_ID = "SampleProducer";
public SampleProducer(String topic, Boolean isAsync) {
final Properties properties = new Properties();
properties.put("bootstrap.servers", KAFKA_SERVER_URL + ":" + KAFKA_SERVER_PORT);
properties.put("client.id", CLIENT_ID);
properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<>(properties);
this.topic = topic;
this.isAsync = isAsync;
}
@Override
public void run() {
int messageNo = 1;
while (true) {
final String messageStr = "Message_" + messageNo;
final long startTime = System.currentTimeMillis();
if (isAsync) { // Send asynchronously
producer.send(new ProducerRecord<>(topic,
messageNo,
messageStr), new DemoCallBack(startTime, messageNo, messageStr));
} else { // Send synchronously
try {
producer.send(new ProducerRecord<>(topic,
messageNo,
messageStr)).get();
System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
// handle the exception
}
}
++messageNo;
}
}
}
class DemoCallBack implements Callback {
private final long startTime;
private final int key;
private final String message;
public DemoCallBack(long startTime, int key, String message) {
this.startTime = startTime;
this.key = key;
this.message = message;
}
/**
* onCompletion method will be called when the record sent to the Kafka Server has been acknowledged.
*
* @param metadata The metadata contains the partition and offset of the record. Null if an error occurred.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
final long elapsedTime = System.currentTimeMillis() - startTime;
if (metadata != null) {
System.out.println(
"message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
"), " +
"offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
} else {
exception.printStackTrace();
}
}
}
public class KafkaProducerDemo {
public static final String TOPIC = "testTopic";
public static void main(String[] args) {
final boolean isAsync = true;
final SampleProducer producerThread = new SampleProducer(TOPIC, isAsync);
// start the producer
producerThread.start();
}
}
When I run KafkaProducerDemo, it prints this at every attempt to send a message to the docker:
[kafka-producer-network-thread | SampleProducer] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=SampleProducer] Connection to node -1 (localhost/127.0.0.1:29092) could not be established. Broker may not be available.
What’s wrong? I’m desperate