I am running a Junit test using EmbeddedKafka,I am able to produce and consume to the embeddedKafka brokers, and successfully assert the data i have send.
But i can see a lot of exceptions in the stack trace, which comes after the assertion is done.
1)
java.io.IOException: Connection to 0 was disconnected before the response was read at org.apache.kafka.clients.NetworkClientUtils.sendAndReceive(NetworkClientUtils.java:97) ~[kafka-clients-2.0.1.jar:na] at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:240) ~[kafka_2.11-2.0.1.jar:na] at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82) [kafka_2.11-2.0.1.jar:na]
2)
2019-10-04 15:49:27.123 WARN 1812 --- [r-0-send-thread] kafka.controller.RequestSendThread : [RequestSendThread controllerId=0] Controller 0's connection to broker localhost:54745 (id: 0 rack: null) was unsuccessful
java.net.SocketTimeoutException: Failed to connect within 1000 ms at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:280) [kafka_2.11-2.0.1.jar:na] at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:233) [kafka_2.11-2.0.1.jar:na] at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82) [kafka_2.11-2.0.1.jar:na]
3)
java.io.IOException: Connection to localhost:54745 (id: 0 rack: null) failed. at org.apache.kafka.clients.NetworkClientUtils.awaitReady(NetworkClientUtils.java:70) ~[kafka-clients-2.0.1.jar:na] at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:279) [kafka_2.11-2.0.1.jar:na] at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:233) [kafka_2.11-2.0.1.jar:na] at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82) [kafka_2.11-2.0.1.jar:na]
4)
java.lang.InterruptedException: null at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(Unknown Source) ~[na:1.8.0_181] at java.util.concurrent.CountDownLatch.await(Unknown Source) ~[na:1.8.0_181] at kafka.utils.ShutdownableThread.pause(ShutdownableThread.scala:69) [kafka_2.11-2.0.1.jar:na] at kafka.controller.RequestSendThread.backoff$1(ControllerChannelManager.scala:221) ~[kafka_2.11-2.0.1.jar:na] at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:235) ~[kafka_2.11-2.0.1.jar:na] at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82) [kafka_2.11-2.0.1.jar:na]
My Test:
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, classes = TestApplication.class)
@ActiveProfiles("test")
@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_CLASS)
//@EmbeddedKafka(controlledShutdown = true)
public class KafkaUtilTest {
private static String TOPIC = "PE";
@Autowired
KafkaUtil kafkaUtil;
@Autowired
ConfigurationProperties configProperties;
SchedulingCallRequestDTO request;
ScheduleOrderResponseDTOv2 response;
Consumer<String, Object> consumer;
HashMap<String, String> expectedHeaderValueMap;
@ClassRule
public static EmbeddedKafkaRule embeddedKafkarule = new EmbeddedKafkaRule(1, true, TOPIC);
@BeforeClass
public static void setUpBeforeClass() throws Exception {
System.setProperty("spring.kafka.producer.bootstrap-servers",
embeddedKafkarule.getEmbeddedKafka().getBrokersAsString());
}
@AfterClass
public static void tearDown() {
embeddedKafkarule.getEmbeddedKafka().getKafkaServers().forEach(b -> b.shutdown());
embeddedKafkarule.getEmbeddedKafka().getKafkaServers().forEach(b -> b.awaitShutdown());
}
@Before
public void init() {
readFile("0211");
expectedHeaderValueMap = getExpectedHeaderValueMap();
Map<String, Object> consumerConfigs = new HashMap<>(
KafkaTestUtils.consumerProps("consumer", "true", embeddedKafkarule.getEmbeddedKafka()));
consumerConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerConfigs.put(JsonDeserializer.TRUSTED_PACKAGES, "com.adessa.promiseengine.dto.kafka");
consumer = new DefaultKafkaConsumerFactory<String, Object>(consumerConfigs).createConsumer();
List<String> topics = new ArrayList<>();
topics.add(TOPIC);
TopicPartition topicPartition1 = new TopicPartition(TOPIC, 0);
TopicPartition topicPartition2 = new TopicPartition(TOPIC, 1);
List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>();
topicPartitions.add(topicPartition1);
topicPartitions.add(topicPartition2);
consumer.assign(topicPartitions);
consumer.seekToBeginning(topicPartitions);
}
@Test
public void testPublish() throws Exception {
kafkaUtil.publishToKafka(request, response);
kafkaUtil.kafkaTemplate.flush();
ConsumerRecord<String, Object> consumedRecord = KafkaTestUtils.getSingleRecord(consumer, TOPIC);
assertRecord(consumedRecord);
}
private void readFile(String testSequenceNo) {}
}
Why are these exceptions there? And during what process they are occuring? Please help
You are using a class rule and the broker is torn down when the test completes, but you are leaving the consumer open; you need to close the consumer at the end of the test.