We are using testcontainers to instantiate Kafka containers for testing an app that creates a kstream StateStore and kafka consumer to process some messages. The test goes fine and we could both produce and consume messages. But these tests continuously print annoying WARN messages -
(kafka-admin-client-thread)([]) WARN - NetworkClient - [AdminClient clientId=test_clientId] Connection to node 1 (localhost/127.0.0.1:55304) could not be established. Broker may not be available.
It seems the AdminClient is scheduled to poll every second and this creates substantial log files. On enabling DEBUG logs, I observe the error
[2022-09-16T22:48:11,747Z](test_x-GlobalStreamThread)([]) DEBUG - NetworkClient - [Consumer clientId=test_x-global-consumer, groupId=null] Node 1 disconnected.
[2022-09-16T22:48:11,747Z](test_x-GlobalStreamThread)([]) WARN - NetworkClient - [Consumer clientId=test_x-global-consumer, groupId=null] Connection to node 1 (localhost/127.0.0.1:55304) could not be established. Broker may not be available.
[2022-09-16T22:48:11,747Z](test_x-GlobalStreamThread)([]) DEBUG - ConsumerNetworkClient - [Consumer clientId=test_x-global-consumer, groupId=null] Cancelled request with header RequestHeader(apiKey=FETCH, apiVersion=11, clientId=test_x-global-consumer, correlationId=11) due to node 1 being disconnected
[2022-09-16T22:48:11,747Z](test_x-GlobalStreamThread)([]) INFO - FetchSessionHandler - [Consumer clientId=test_x-global-consumer, groupId=null] Error sending fetch request (sessionId=1421844635, epoch=INITIAL) to node 1:
org.apache.kafka.common.errors.DisconnectException: null
[2022-09-16T22:48:11,792Z](kafka-producer-network-thread | producer-1)([]) DEBUG - NetworkClient - [Producer clientId=producer-1] Initialize connection to node localhost:55304 (id: 1 rack: null) for sending metadata request
[2022-09-16T22:48:11,792Z](kafka-admin-client-thread | adminclient-1)([]) DEBUG - NetworkClient - [AdminClient clientId=adminclient-1] Initiating connection to node localhost:55304 (id: 1 rack: null) using address localhost/127.0.0.1
[2022-09-16T22:48:11,792Z](kafka-producer-network-thread | producer-1)([]) DEBUG - NetworkClient - [Producer clientId=producer-1] Initiating connection to node localhost:55304 (id: 1 rack: null) using address localhost/127.0.0.1
[2022-09-16T22:48:11,793Z](kafka-producer-network-thread | producer-1)([]) DEBUG - Selector - [Producer clientId=producer-1] Connection with localhost/127.0.0.1 disconnected
java.net.ConnectException: Connection refused
at java.base/sun.nio.ch.Net.pollConnect(Native Method)
at java.base/sun.nio.ch.Net.pollConnectNow(Net.java:672)
at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:946)
at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:219)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:530)
at org.apache.kafka.common.network.Selector.poll(Selector.java:485)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
at java.base/java.lang.Thread.run(Thread.java:833)
[2022-09-16T22:48:11,793Z](kafka-admin-client-thread | adminclient-1)([]) DEBUG - Selector - [AdminClient clientId=adminclient-1] Connection with localhost/127.0.0.1 disconnected
java.net.ConnectException: Connection refused
at java.base/sun.nio.ch.Net.pollConnect(Native Method)
at java.base/sun.nio.ch.Net.pollConnectNow(Net.java:672)
at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:946)
at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:219)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:530)
at org.apache.kafka.common.network.Selector.poll(Selector.java:485)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)
at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1293)
at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1224)
at java.base/java.lang.Thread.run(Thread.java:833)
[2022-09-16T22:48:11,793Z](kafka-producer-network-thread | producer-1)([]) DEBUG - NetworkClient - [Producer clientId=producer-1] Node 1 disconnected.
[2022-09-16T22:48:11,793Z](kafka-admin-client-thread | adminclient-1)([]) DEBUG - NetworkClient - [AdminClient clientId=adminclient-1] Node 1 disconnected.Similar logs are also coming from GlobalStreamThread
[2022-09-16T22:48:11,746Z](test_x-GlobalStreamThread)([]) DEBUG - NetworkClient - [Consumer clientId=test_x-global-consumer, groupId=null] Initiating connection to node localhost:55304 (id: 1 rack: null) using address localhost/127.0.0.1
[2022-09-16T22:48:11,747Z](test_x-GlobalStreamThread)([]) DEBUG - Selector - [Consumer clientId=test_x-global-consumer, groupId=null] Connection with localhost/127.0.0.1 disconnected
java.net.ConnectException: Connection refused
at java.base/sun.nio.ch.Net.pollConnect(Native Method)
at java.base/sun.nio.ch.Net.pollConnectNow(Net.java:672)
at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:946)
at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:219)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:530)
at org.apache.kafka.common.network.Selector.poll(Selector.java:485)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1296)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)
at org.apache.kafka.streams.processor.internals.GlobalStreamThread$StateConsumer.pollAndUpdate(GlobalStreamThread.java:237)
at org.apache.kafka.streams.processor.internals.GlobalStreamThread.run(GlobalStreamThread.java:284)Docker port mapping looks like
0.0.0.0:55306->2181/tcp, 0.0.0.0:55305->9092/tcp, 0.0.0.0:55304->9093/tcp
Testcontainer is started as
KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1")) .withNetwork(Network.SHARED)
Testcontainer maps KAFKA_LISTENERS here https://github.com/testcontainers/testcontainers-java/blob/master/modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java#L49
Could you help me understand what is happening here and how to provide the correct port mappings via testcontainers for these threads to work fine?