From 0fb431db3ae326cb6d518c080ceac6297cc8d4c4 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Tue, 9 Feb 2016 18:17:41 +0530 Subject: [PATCH] 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 --- .../kafka/KafkaBinderHealthIndicator.java | 88 +++++++++++++++++++ .../kafka/KafkaMessageChannelBinder.java | 69 ++++++++------- .../KafkaBinderConfigurationProperties.java | 26 ++++++ .../config/KafkaServiceAutoConfiguration.java | 11 ++- .../kafka-binder.properties | 2 + 5 files changed, 165 insertions(+), 31 deletions(-) create mode 100644 spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java new file mode 100644 index 000000000..b07986fa6 --- /dev/null +++ b/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 brokersInClusterSet = new HashSet<>(); + try { + Seq allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient); + Collection 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 downMessages = new HashSet<>(); + for (Map.Entry> 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(); + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index d1297a196..132a1b140 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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> topicsInUse = new HashMap<>(); protected static final Set PRODUCER_COMPRESSION_PROPERTIES = new HashSet( 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> getTopicsInUse() { + return this.topicsInUse; + } + @Override protected Binding 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 partitions = ensureTopicCreated(name, numPartitions, defaultReplicationFactor); + topicsInUse.put(name, partitions); + ProducerMetadata 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 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) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index 99f63f00e..a03fc1cde 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/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. * diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java index fb94c1858..84bd9c08e 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaServiceAutoConfiguration.java +++ b/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); + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties index 41e6ea09f..ef3773e0f 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/kafka-binder.properties +++ b/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