GH-231: reuse Kafka consumer HealthIndicator

- lazily create Consumer in KafkaBinderHealthIndicator.

- polishing

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java

@@ -41,6 +41,7 @@ import org.springframework.kafka.core.ConsumerFactory;
  * @author Marius Bogoevici
  * @author Henryk Konsek
  * @author Gary Russell
+ * @author Laur Aliste
  */
 public class KafkaBinderHealthIndicator implements HealthIndicator {
 
@@ -52,15 +53,16 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
 
 	private int timeout = DEFAULT_TIMEOUT;
 
-	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
-			ConsumerFactory<?, ?> consumerFactory) {
+	private Consumer<?, ?> metadataConsumer;
+
+	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder, ConsumerFactory<?, ?> consumerFactory) {
 		this.binder = binder;
 		this.consumerFactory = consumerFactory;
-
 	}
 
 	/**
 	 * Set the timeout in seconds to retrieve health information.
+	 *
 	 * @param timeout the timeout - default 60.
 	 */
 	public void setTimeout(int timeout) {
@@ -74,7 +76,10 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
 
 			@Override
 			public Health call() {
-				try (Consumer<?, ?> metadataConsumer = consumerFactory.createConsumer()) {
+				try {
+					if (metadataConsumer == null) {
+						metadataConsumer = consumerFactory.createConsumer();
+					}
 					Set<String> downMessages = new HashSet<>();
 					for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) {
 						List<PartitionInfo> partitionInfos = metadataConsumer.partitionsFor(topic);

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java

@@ -22,11 +22,13 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -40,6 +42,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * @author Barry Commins
  * @author Gary Russell
+ * @author Laur Aliste
  */
 public class KafkaBinderHealthIndicatorTest {
 
@@ -74,7 +77,6 @@ public class KafkaBinderHealthIndicatorTest {
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
 		Health health = indicator.health();
 		assertThat(health.getStatus()).isEqualTo(Status.UP);
-		org.mockito.Mockito.verify(this.consumer).close();
 	}
 
 	@Test
@@ -105,6 +107,33 @@ public class KafkaBinderHealthIndicatorTest {
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
 	}
 
+	@Test
+	public void createsConsumerOnceWhenInvokedMultipleTimes() {
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group", partitions));
+		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions);
+
+		indicator.health();
+		Health health = indicator.health();
+
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+		org.mockito.Mockito.verify(this.consumerFactory).createConsumer();
+	}
+
+	@Test
+	public void consumerCreationFailsFirstTime() {
+		org.mockito.BDDMockito.given(consumerFactory.createConsumer()).willThrow(KafkaException.class)
+				.willReturn(consumer);
+
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
+
+		health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+
+		org.mockito.Mockito.verify(this.consumerFactory, Mockito.times(2)).createConsumer();
+	}
+
 	private List<PartitionInfo> partitions(Node leader) {
 		List<PartitionInfo> partitions = new ArrayList<>();
 		partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null));