SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Corrections for health indicator

Browse Source 
Ensure that try-catch is applied around connecting to Zookeeper as well
Register topics in use for consumer
This commit is contained in:
Marius Bogoevici
2016-02-19 16:48:33 -05:00
parent 0fb431db3a
commit b1b0e960d7
2 changed files with 18 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -48,15 +48,29 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
@Override
public Health health() {
ZkClient zkClient = new ZkClient(binder.getZkAddress(), binder.getZkSessionTimeout(),
binder.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$);
Set<String> brokersInClusterSet = new HashSet<>();
ZkClient zkClient = null;
try {
zkClient = new ZkClient(binder.getZkAddress(), binder.getZkSessionTimeout(),
binder.getZkConnectionTimeout(), ZKStringSerializer$.MODULE$);
Set<String> brokersInClusterSet = new HashSet<>();
Seq<Broker> allBrokersInCluster = ZkUtils$.MODULE$.getAllBrokersInCluster(zkClient);
Collection<Broker> brokersInCluster = JavaConversions.asJavaCollection(allBrokersInCluster);
for (Broker broker : brokersInCluster) {
brokersInClusterSet.add(broker.connectionString());
}
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();
}
catch (Exception e) {
return Health.down(e).build();
@@ -71,18 +85,5 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
}
}
}
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();
}
}

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

@@ -600,7 +600,7 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
}
}
}
topicsInUse.put(name, listenedPartitions);
ReceivingHandler rh = new ReceivingHandler();
rh.setOutputChannel(moduleInputChannel);