Add junit test for health indicator

pom.xml

@@ -23,10 +23,12 @@
 		<docker-maven-plugin.version>0.27.2</docker-maven-plugin.version>
 		<jgitflow-maven-plugin.version>1.0-m5.1</jgitflow-maven-plugin.version>
 		<spring-boot-starter.version>2.1.1.RELEASE</spring-boot-starter.version>
-		<spring.kafka.version>2.1.7.RELEASE</spring.kafka.version>
+		<spring.kafka.version>2.2.4.RELEASE</spring.kafka.version>
 
 		<maven-surefire-plugin.version>2.22.1</maven-surefire-plugin.version>
 		<maven-failsafe-plugin.version>2.22.1</maven-failsafe-plugin.version>
+		<assertj-core.version>3.11.1</assertj-core.version>
+		<guava.version>27.1-jre</guava.version>
 	</properties>
 
 	<dependencies>
@@ -42,6 +44,11 @@
 			<version>${spring.kafka.version}</version>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<version>${guava.version}</version>
+		</dependency>
 
 		<dependency>
 			<groupId>org.junit.jupiter</groupId>
@@ -61,6 +68,24 @@
 			<version>${junit.jupiter.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.assertj</groupId>
+			<artifactId>assertj-core</artifactId>
+			<version>${assertj-core.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-test</artifactId>
+			<version>2.1.2.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<version>${spring.kafka.version}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>

src/main/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicator.java

@@ -2,6 +2,7 @@ package com.deviceinsight.kafka.health;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -19,6 +20,7 @@ import org.springframework.boot.actuate.health.Health;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -100,7 +102,8 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
 		}
 	}
 
-	private void subscribeToTopic() throws InterruptedException {
+	@VisibleForTesting
+	void subscribeToTopic() throws InterruptedException {
 
 		final CountDownLatch subscribed = new CountDownLatch(1);
 
@@ -123,7 +126,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
 			}
 		});
 
-		consumer.poll(pollTimeoutMs);
+		consumer.poll(Duration.ofMillis(pollTimeoutMs));
 		if (!subscribed.await(subscriptionTimeoutMs, MILLISECONDS)) {
 			throw new RuntimeException("Subscription to kafka failed, topic=" + topic);
 		}
@@ -173,7 +176,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
 
 	private boolean messageNotReceived(String message) {
 
-		return StreamSupport.stream(consumer.poll(pollTimeoutMs).spliterator(), false)
+		return StreamSupport.stream(consumer.poll(Duration.ofMillis(pollTimeoutMs)).spliterator(), false)
 				.noneMatch(msg -> msg.key().equals(message) && msg.value().equals(message));
 	}
 

src/test/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicatorTest.java

@@ -0,0 +1,82 @@
+package com.deviceinsight.kafka.health;
+
+import static com.deviceinsight.kafka.health.KafkaConsumingHealthIndicatorTest.TOPIC;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import kafka.server.KafkaServer;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.core.BrokerAddress;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@ExtendWith(SpringExtension.class)
+@EmbeddedKafka(topics = {TOPIC})
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class KafkaConsumingHealthIndicatorTest {
+
+	static final String TOPIC = "health-checks";
+
+	private Consumer<String, String> consumer;
+
+	@Autowired
+	private EmbeddedKafkaBroker embeddedKafkaBroker;
+
+	@BeforeAll
+	public void setUp() {
+		Map<String, Object> consumerConfigs =
+				new HashMap<>(KafkaTestUtils.consumerProps("consumer", "false", embeddedKafkaBroker));
+		consumer = new DefaultKafkaConsumerFactory<>(consumerConfigs, new StringDeserializer(),
+				new StringDeserializer()).createConsumer();
+		consumer.subscribe(Collections.singletonList(TOPIC));
+		consumer.poll(Duration.ofSeconds(1));
+	}
+
+	@AfterAll
+	public void tearDown() {
+		consumer.close();
+		embeddedKafkaBroker.getKafkaServers().forEach(KafkaServer::shutdown);
+		embeddedKafkaBroker.getKafkaServers().forEach(KafkaServer::awaitShutdown);
+	}
+
+	@Test
+	public void kafkaIsDown() throws Exception {
+		KafkaHealthProperties kafkaHealthProperties = new KafkaHealthProperties();
+		kafkaHealthProperties.setTopic(TOPIC);
+
+		final KafkaProperties kafkaProperties = new KafkaProperties();
+		BrokerAddress[] brokerAddresses = embeddedKafkaBroker.getBrokerAddresses();
+		kafkaProperties.setBootstrapServers(Collections.singletonList(brokerAddresses[0].toString()));
+
+		KafkaConsumingHealthIndicator healthIndicator =
+				new KafkaConsumingHealthIndicator(kafkaHealthProperties, kafkaProperties.buildConsumerProperties(),
+						kafkaProperties.buildProducerProperties());
+		healthIndicator.subscribeToTopic();
+
+		Health health = healthIndicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+
+		shutdownKafka();
+	}
+
+	private void shutdownKafka() {
+		this.embeddedKafkaBroker.destroy();
+	}
+}