Add Kafka binder health indicator

- Health indicator fetches broker addresses using ZK configuration and matches it with the leaders of the paritions being used in the binder

This resolves #297

Use Set instead of List to store error messages per broker

Exception handling when connecting to ZK

Add ZK connect/session timeout values as configuration properties

 - Set it in the Kafka binder so that the same can be used for health indicator as well

Rename ZK properties and move them to binder configuration properties

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

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.I0Itec.zkclient.ZkClient;
+
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.integration.kafka.core.BrokerAddress;
+import org.springframework.integration.kafka.core.Partition;
+
+import kafka.cluster.Broker;
+import kafka.utils.ZKStringSerializer$;
+import kafka.utils.ZkUtils$;
+import scala.collection.JavaConversions;
+import scala.collection.Seq;
+
+/**
+ * Health indicator for Kafka.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+public class KafkaBinderHealthIndicator implements HealthIndicator {
+
+	private final KafkaMessageChannelBinder binder;
+
+	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder) {
+		this.binder = binder;
+	}
+
+	@Override
+	public Health health() {
+		ZkClient zkClient = new ZkClient(binder.getZkAddress(), binder.getZkSessionTimeout(),
+				binder.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$);
+		Set<String> brokersInClusterSet = new HashSet<>();
+		try {
+			Seq<Broker> allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient);
+			Collection<Broker> brokersInCluster = JavaConversions.asJavaCollection(allBrokersInCluster);
+			for (Broker broker : brokersInCluster) {
+				brokersInClusterSet.add(broker.connectionString());
+			}
+		}
+		catch (Exception e) {
+			return Health.down(e).build();
+		}
+		finally {
+			if (zkClient != null) {
+				try {
+					zkClient.close();
+				}
+				catch (Exception e) {
+					// ignore
+				}
+			}
+		}
+		Set<String> downMessages = new HashSet<>();
+		for (Map.Entry<String, Collection<Partition>> entry : binder.getTopicsInUse().entrySet()) {
+			for (Partition partition : entry.getValue()) {
+				BrokerAddress address = binder.getConnectionFactory().getLeader(partition);
+				if (!brokersInClusterSet.contains(address.toString())) {
+					downMessages.add(address.toString());
+				}
+			}
+		}
+		if (downMessages.isEmpty()) {
+			return Health.up().build();
+		}
+		return Health.down().withDetail("Following brokers are down: ", downMessages.toString()).build();
+	}
+}

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

@@ -21,6 +21,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -31,8 +32,6 @@ import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.I0Itec.zkclient.ZkClient;
-import org.I0Itec.zkclient.exception.ZkMarshallingError;
-import org.I0Itec.zkclient.serialize.ZkSerializer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 
@@ -85,6 +84,7 @@ import kafka.admin.AdminUtils;
 import kafka.api.OffsetRequest;
 import kafka.serializer.Decoder;
 import kafka.serializer.DefaultDecoder;
+import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
 import scala.collection.Seq;
 
@@ -157,35 +157,15 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 
 	private static final boolean DEFAULT_RESET_OFFSETS = false;
 
+	private static final int DEFAULT_ZK_SESSION_TIMEOUT = 10000;
+
+	private static final int DEFAULT_ZK_CONNECTION_TIMEOUT = 10000;
+
 	private static final StartOffset DEFAULT_START_OFFSET = StartOffset.latest;
 
 	private RetryOperations retryOperations;
 
-	/**
-	 * Used when writing directly to ZK. This is what Kafka expects.
-	 */
-	public final static ZkSerializer utf8Serializer = new ZkSerializer() {
-
-		@Override
-		public byte[] serialize(Object data) throws ZkMarshallingError {
-			try {
-				return ((String) data).getBytes("UTF-8");
-			}
-			catch (UnsupportedEncodingException e) {
-				throw new ZkMarshallingError(e);
-			}
-		}
-
-		@Override
-		public Object deserialize(byte[] bytes) throws ZkMarshallingError {
-			try {
-				return new String(bytes, "UTF-8");
-			}
-			catch (UnsupportedEncodingException e) {
-				throw new ZkMarshallingError(e);
-			}
-		}
-	};
+	private Map<String, Collection<Partition>> topicsInUse = new HashMap<>();
 
 	protected static final Set<Object> PRODUCER_COMPRESSION_PROPERTIES = new HashSet<Object>(
 			Arrays.asList(new String[] {
@@ -270,6 +250,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 
 	private StartOffset startOffset = DEFAULT_START_OFFSET;
 
+	private int zkSessionTimeout = DEFAULT_ZK_SESSION_TIMEOUT;
+
+	private int zkConnectionTimeout = DEFAULT_ZK_CONNECTION_TIMEOUT;
+
 	private ProducerListener producerListener;
 
 	public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress,
@@ -290,6 +274,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 		}
 	}
 
+	String getZkAddress() {
+		return this.zkAddress;
+	}
+
 	public void setSocketBufferSize(int socketBufferSize) {
 		this.socketBufferSize = socketBufferSize;
 	}
@@ -424,6 +412,26 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 		this.startOffset = startOffset;
 	}
 
+	public int getZkSessionTimeout() {
+		return this.zkSessionTimeout;
+	}
+
+	public void setZkSessionTimeout(int zkSessionTimeout) {
+		this.zkSessionTimeout = zkSessionTimeout;
+	}
+
+	public int getZkConnectionTimeout() {
+		return this.zkConnectionTimeout;
+	}
+
+	public void setZkConnectionTimeout(int zkConnectionTimeout) {
+		this.zkConnectionTimeout = zkConnectionTimeout;
+	}
+
+	Map<String, Collection<Partition>> getTopicsInUse() {
+		return this.topicsInUse;
+	}
+
 	@Override
 	protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) {
 		// If the caller provides a consumer group, use it; otherwise an anonymous consumer group
@@ -452,6 +460,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 
 		Collection<Partition> partitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor);
 
+		topicsInUse.put(name, partitions);
+
 		ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<>(
 				name, byte[].class, byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
 		producerMetadata.setCompressionType(ProducerMetadata.CompressionType.valueOf(
@@ -496,9 +506,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
 	private Collection<Partition> ensureTopicCreated(final String topicName, final int numPartitions,
 			int replicationFactor) {
 
-		final int sessionTimeoutMs = 10000;
-		final int connectionTimeoutMs = 10000;
-		final ZkClient zkClient = new ZkClient(zkAddress, sessionTimeoutMs, connectionTimeoutMs, utf8Serializer);
+		final ZkClient zkClient = new ZkClient(zkAddress, getZkSessionTimeout(), getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
 		try {
 			// The following is basically copy/paste from AdminUtils.createTopic() with
 			// createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true)

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java

@@ -51,6 +51,16 @@ class KafkaBinderConfigurationProperties {
 
 	private KafkaMessageChannelBinder.StartOffset startOffset;
 
+	/**
+	 * ZK session timeout in milliseconds.
+	 */
+	private int zkSessionTimeout;
+
+	/**
+	 * ZK Connection timeout in milliseconds.
+	 */
+	private int zkConnectionTimeout;
+
 	public String getZkConnectionString() {
 		return toConnectionString(this.zkNodes, this.defaultZkPort);
 	}
@@ -132,6 +142,22 @@ class KafkaBinderConfigurationProperties {
 		this.resetOffsets = resetOffsets;
 	}
 
+	public int getZkSessionTimeout() {
+		return this.zkSessionTimeout;
+	}
+
+	public void setZkSessionTimeout(int zkSessionTimeout) {
+		this.zkSessionTimeout = zkSessionTimeout;
+	}
+
+	public int getZkConnectionTimeout() {
+		return this.zkConnectionTimeout;
+	}
+
+	public void setZkConnectionTimeout(int zkConnectionTimeout) {
+		this.zkConnectionTimeout = zkConnectionTimeout;
+	}
+
 	/**
 	 * Converts an array of host values to a comma-separated String.
 	 *

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java

@@ -21,6 +21,7 @@ import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfigurati
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
 import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
 import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
 import org.springframework.context.annotation.Bean;
@@ -74,7 +75,7 @@ public class KafkaServiceAutoConfiguration {
 		KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
 				new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString)
 				: new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString,
-						headers);
+				headers);
 		kafkaMessageChannelBinder.setCodec(codec);
 		kafkaMessageChannelBinder.setMode(kafkaBinderConfigurationProperties.getMode());
 		kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(kafkaBinderConfigurationProperties.getOffsetUpdateTimeWindow());
@@ -84,6 +85,9 @@ public class KafkaServiceAutoConfiguration {
 		kafkaMessageChannelBinder.setResetOffsets(kafkaBinderConfigurationProperties.isResetOffsets());
 		kafkaMessageChannelBinder.setStartOffset(kafkaBinderConfigurationProperties.getStartOffset());
 
+		kafkaMessageChannelBinder.setZkSessionTimeout(kafkaBinderConfigurationProperties.getZkSessionTimeout());
+		kafkaMessageChannelBinder.setZkConnectionTimeout(kafkaBinderConfigurationProperties.getZkConnectionTimeout());
+
 		kafkaMessageChannelBinder.setDefaultAutoCommitEnabled(kafkaBinderDefaultProperties.isAutoCommitEnabled());
 		kafkaMessageChannelBinder.setDefaultBatchSize(kafkaBinderDefaultProperties.getBatchSize());
 		kafkaMessageChannelBinder.setDefaultBatchTimeout(kafkaBinderDefaultProperties.getBatchTimeout());
@@ -104,4 +108,9 @@ public class KafkaServiceAutoConfiguration {
 	ProducerListener producerListener() {
 		return new LoggingProducerListener();
 	}
+
+	@Bean
+	KafkaBinderHealthIndicator healthIndicator(KafkaMessageChannelBinder kafkaMessageChannelBinder) {
+		return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder);
+	}
 }

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties

@@ -6,6 +6,8 @@ spring.cloud.stream.binder.kafka.mode=embeddedHeaders
 spring.cloud.stream.binder.kafka.offsetUpdateTimeWindow=10000
 spring.cloud.stream.binder.kafka.offsetUpdateCount=0
 spring.cloud.stream.binder.kafka.offsetUpdateShutdownTimeout=2000
+spring.cloud.stream.binder.kafka.zkSessionTimeout=10000
+spring.cloud.stream.binder.kafka.zkConnectionTimeout=10000
 spring.cloud.stream.binder.kafka.default.batchSize=16384
 spring.cloud.stream.binder.kafka.default.batchTimeout=0
 spring.cloud.stream.binder.kafka.default.requiredAcks=1