Confluent Platform unable to properly send/consume message into the Kafka topic

Question

Currently, I use Maven with io.fabric8 docker-maven-plugin in order to automatically spin up Kafka and ZooKeeper. This is my current configuration which works fine:

<properties>
    <zookeeper.port>2181</zookeeper.port>

    <kafka.host>127.0.0.1</kafka.host>
    <kafka.port>9092</kafka.port>
</properties>

<plugin>
    <groupId>io.fabric8</groupId>
    <artifactId>docker-maven-plugin</artifactId>
    <version>${docker-maven-plugin.version}</version>

    <configuration>
        <showLogs>true</showLogs>
        <images>

            <image>
                <name>wurstmeister/zookeeper:${zookeeper.version}</name>
                <alias>zookeeper</alias>
                <run>
                    <ports>
                        <port>${zookeeper.port}:2181</port>
                    </ports>
                </run>
            </image>

            <image>
                <name>wurstmeister/kafka:${kafka.version}</name>
                <alias>kafka</alias>
                <run>
                    <ports>
                        <port>${kafka.port}:9092</port>
                    </ports>
                    <links>
                        <link>zookeeper:zookeeper</link>
                    </links>
                    <env>
                        <KAFKA_ADVERTISED_HOST_NAME>${local.ip}
                        </KAFKA_ADVERTISED_HOST_NAME>
                        <KAFKA_ADVERTISED_PORT>${kafka.port}</KAFKA_ADVERTISED_PORT>
                        <KAFKA_ZOOKEEPER_CONNECT>zookeeper:2181</KAFKA_ZOOKEEPER_CONNECT>
                        <KAFKA_MESSAGE_MAX_BYTES>15000000</KAFKA_MESSAGE_MAX_BYTES>
                    </env>
                </run>
            </image>

            <image>
                <name>confluentinc/cp-ksql-server:5.0.0</name>
                <alias>cp-ksql-server</alias>
                <run>
                    <ports>
                        <port>8088:8088</port>
                    </ports>
                    <links>
                        <link>kafka:kafka</link>
                    </links>
                    <env>
                        <KSQL_BOOTSTRAP_SERVERS>${local.ip}:9092</KSQL_BOOTSTRAP_SERVERS>
                        <KSQL_LISTENERS>http://0.0.0.0:8088/</KSQL_LISTENERS>
                        <KSQL_KSQL_SERVICE_ID>confluent_test_2</KSQL_KSQL_SERVICE_ID>
                    </env>
                </run>
            </image>

        </images>

    </configuration>
</plugin>

Right now I need more flexibility and monitoring tools in order to be able to properly analyze my local Kafka instance and this is why I decide to start using All-In-One Confluent Platform https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html

I performed the steps described in the documentation above and was able to run Confluent Platform via docker-compose up -d command on the following docker-compose.yml https://github.com/confluentinc/cp-docker-images/tree/master/examples/cp-all-in-one. After that I was able to access Confluent Control Center on http://localhost:9021

My application is also starts but messages are not sending or consumed properly. This is the log:

2018-08-25 20:54:33.786  INFO 11304 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.1.0
2018-08-25 20:54:33.786  INFO 11304 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : fdcf75ea326b8e07
2018-08-25 20:54:33.794  INFO 11304 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : Cluster ID: NCbZ7abTTdu6SQHhthMbIw
2018-08-25 20:54:33.827  INFO 11304 --- [nio-8080-exec-1] c.p.domain.service.post.PostServiceImpl  : Message sent to the post creation queue: Post [id=5b8197d9e67db62c28ebe8d5, status=PENDING, externalPostId=2bdbd597-cc30-4478-b946-e873c8d53eec, chatName= ....]

2018-08-25 20:55:03.875 ERROR 11304 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='Post [id=5b8197d9e67db62c28ebe8d5, status=PENDING, externalPostId=2bdbd597-cc30-4478-b946-e873c8d53e...' to topic post.send:

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for post.send-0: 30047 ms has passed since batch creation plus linger time

I can see my topics there:

What may be wrong there and how to fix it ?

UPDATED

I enabled the DEBUG logs on Kafka in my application and right now can see the following error:

java.io.IOException: Can't resolve address: broker:9092
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:235) ~[kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.common.network.Selector.connect(Selector.java:214) ~[kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:793) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:230) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:792) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1002) [kafka-clients-1.1.0.jar:na]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101) ~[na:1.8.0_171]
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[na:1.8.0_171]
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:233) ~[kafka-clients-1.1.0.jar:na]
    ... 6 common frames omitted

Not sure why you're showing the Maven piece if you're just using Docker Compose. You could add the monitoring interceptors to the wurstmeister images, however that is more work. — OneCricketeer
Just to show the previous configuration where the logic works fine. May be it will help to better understand the issue with the Confluent stack. Right now I have no idea what may be wrong there — alexanoid
"Expiring x record(s) for topic-partition: XX ms has passed since batch creation" is a issue with either the broker or the producer. Are you seeing any logs from either that would indicate an issue? How many messages are you sending at once? Kafka Producer will batch records, so you need to send maybe 100, or at the very least set a while loop — OneCricketeer
I just use absolutely the same scenario I used for my Maven configuration - I send only one test message. With Maven config everything is working fine, with Confluent stack - the mentioned issue appears. This is the Kafka logs from Docker container - files.fm/u/8b98cpt6 — alexanoid
Thank you very much !!! Right now everything is working like a charm! — alexanoid

OneCricketeer OneCricketeer · Accepted Answer · 2018-08-25T21:38:02

Can't resolve address: broker:9092

Kafka is returning back to the clients the listeners that it knows about

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092

Basically, bootstrap.servers only needs at least one address, but the rest of the Kafka cluster addresses will be returned back to the client in order to establish connections to individual brokers.

When you connect outside of the Docker network to localhost:9092, then the Docker port forward will work, but then Kafka returns back to your client broker:9092, and the client disconnects because you've not setup DNS between your host and Docker network (and you shouldn't need to unless using Docker Swarm).

Your solution would be two options

Use localhost:29092 in your code, and things might start working better
Run your Java code in a Docker container on the same network setup by the Compose file and leave the kafka:9092 reference (this is similar to how your KSQL and Control Center containers would be working)

Confluent Platform unable to properly send/consume message into the Kafka topic

1 Answers