bvn13/kafka-health-check
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

change health properties to durations

Browse Source
This commit is contained in:
Simon Flandergan 2019-10-10 20:33:20 +02:00
parent c4706f9403
commit 06d4941363
4 changed files with 46 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
changelog.adoc
View File

@ -1,6 +1,10 @@
= KafkaHealthCheck = KafkaHealthCheck
:icons: font :icons: font
== Version 2.0.0
* Changed properties to duration type
== Version 1.1.0 == Version 1.1.0
* Make consumer groups unique by appending a random UUID when no group ID is configured explicitly. * Make consumer groups unique by appending a random UUID when no group ID is configured explicitly.

3
pom.xml
View File

@ -6,7 +6,7 @@
<groupId>com.deviceinsight.kafka</groupId> <groupId>com.deviceinsight.kafka</groupId>
<artifactId>kafka-health-check</artifactId> <artifactId>kafka-health-check</artifactId>
<version>1.3.0-SNAPSHOT</version> <version>2.0.0-SNAPSHOT</version>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>Kafka Health Check</name> <name>Kafka Health Check</name>
@ -36,6 +36,7 @@
<maven-gpg-plugin.version>1.6</maven-gpg-plugin.version> <maven-gpg-plugin.version>1.6</maven-gpg-plugin.version>
<maven-javadoc-plugin.version>3.1.0</maven-javadoc-plugin.version> <maven-javadoc-plugin.version>3.1.0</maven-javadoc-plugin.version>
<gitflow-maven-plugin.version>1.12.0</gitflow-maven-plugin.version> <gitflow-maven-plugin.version>1.12.0</gitflow-maven-plugin.version>
<lombok.version>1.18.10</lombok.version>
</properties> </properties>
<dependencies> <dependencies>

35
src/main/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicator.java
View File

@ -51,9 +51,9 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
private final Producer<String, String> producer; private final Producer<String, String> producer;
private final String topic; private final String topic;
private final long sendReceiveTimeoutMs; private final Duration sendReceiveTimeout;
private final long pollTimeoutMs; private final Duration pollTimeout;
private final long subscriptionTimeoutMs; private final Duration subscriptionTimeout;
private final ExecutorService executor; private final ExecutorService executor;
private final AtomicBoolean running; private final AtomicBoolean running;
@ -64,10 +64,11 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
public KafkaConsumingHealthIndicator(KafkaHealthProperties kafkaHealthProperties, public KafkaConsumingHealthIndicator(KafkaHealthProperties kafkaHealthProperties,
Map<String, Object> kafkaConsumerProperties, Map<String, Object> kafkaProducerProperties) { Map<String, Object> kafkaConsumerProperties, Map<String, Object> kafkaProducerProperties) {
logger.info("Initializing kafka health check with properties: {}", kafkaHealthProperties);
this.topic = kafkaHealthProperties.getTopic(); this.topic = kafkaHealthProperties.getTopic();
this.sendReceiveTimeoutMs = kafkaHealthProperties.getSendReceiveTimeoutMs(); this.sendReceiveTimeout = kafkaHealthProperties.getSendReceiveTimeout();
this.pollTimeoutMs = kafkaHealthProperties.getPollTimeoutMs(); this.pollTimeout = kafkaHealthProperties.getPollTimeout();
this.subscriptionTimeoutMs = kafkaHealthProperties.getSubscriptionTimeoutMs(); this.subscriptionTimeout = kafkaHealthProperties.getSubscriptionTimeout();
Map<String, Object> kafkaConsumerPropertiesCopy = new HashMap<>(kafkaConsumerProperties); Map<String, Object> kafkaConsumerPropertiesCopy = new HashMap<>(kafkaConsumerProperties);
@ -81,7 +82,8 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
this.executor = Executors.newSingleThreadExecutor(); this.executor = Executors.newSingleThreadExecutor();
this.running = new AtomicBoolean(true); this.running = new AtomicBoolean(true);
this.cache = Caffeine.newBuilder().expireAfterWrite(sendReceiveTimeoutMs, TimeUnit.MILLISECONDS).build(); this.cache =
Caffeine.newBuilder().expireAfterWrite(sendReceiveTimeout.toMillis(), TimeUnit.MILLISECONDS).build();
this.kafkaCommunicationResult = this.kafkaCommunicationResult =
KafkaCommunicationResult.failure(new RejectedExecutionException("Kafka Health Check is starting.")); KafkaCommunicationResult.failure(new RejectedExecutionException("Kafka Health Check is starting."));
@ -92,12 +94,13 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
subscribeToTopic(); subscribeToTopic();
if (kafkaCommunicationResult.isFailure()) { if (kafkaCommunicationResult.isFailure()) {
throw new BeanInitializationException("Kafka health check failed", kafkaCommunicationResult.getException()); throw new BeanInitializationException("Kafka health check failed",
kafkaCommunicationResult.getException());
} }
executor.submit(() -> { executor.submit(() -> {
while (running.get()) { while (running.get()) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(pollTimeoutMs)); ConsumerRecords<String, String> records = consumer.poll(pollTimeout);
records.forEach(record -> cache.put(record.key(), record.value())); records.forEach(record -> cache.put(record.key(), record.value()));
} }
}); });
@ -145,8 +148,8 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
} }
}); });
consumer.poll(Duration.ofMillis(pollTimeoutMs)); consumer.poll(pollTimeout);
if (!subscribed.await(subscriptionTimeoutMs, MILLISECONDS)) { if (!subscribed.await(subscriptionTimeout.toMillis(), MILLISECONDS)) {
throw new BeanInitializationException("Subscription to kafka failed, topic=" + topic); throw new BeanInitializationException("Subscription to kafka failed, topic=" + topic);
} }
@ -157,7 +160,6 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
try { try {
return sendKafkaMessage(); return sendKafkaMessage();
} catch (ExecutionException e) { } catch (ExecutionException e) {
logger.warn("Kafka health check execution failed.", e); logger.warn("Kafka health check execution failed.", e);
this.kafkaCommunicationResult = KafkaCommunicationResult.failure(e); this.kafkaCommunicationResult = KafkaCommunicationResult.failure(e);
@ -177,7 +179,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
logger.trace("Send health check message = {}", message); logger.trace("Send health check message = {}", message);
producer.send(new ProducerRecord<>(topic, message, message)).get(sendReceiveTimeoutMs, MILLISECONDS); producer.send(new ProducerRecord<>(topic, message, message)).get(sendReceiveTimeout.toMillis(), MILLISECONDS);
return message; return message;
} }
@ -197,25 +199,22 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
builder.up(); builder.up();
return; return;
} else if (System.currentTimeMillis() - startTime > sendReceiveTimeout.toMillis()) {
} else if (System.currentTimeMillis() - startTime > sendReceiveTimeoutMs) {
if (kafkaCommunicationResult.isFailure()) { if (kafkaCommunicationResult.isFailure()) {
goDown(builder); goDown(builder);
} else { } else {
builder.down(new TimeoutException( builder.down(new TimeoutException(
"Sending and receiving took longer than " + sendReceiveTimeoutMs + " ms")) "Sending and receiving took longer than " + sendReceiveTimeout + " ms"))
.withDetail("topic", topic); .withDetail("topic", topic);
} }
return; return;
} }
} }
} }
private void goDown(Health.Builder builder) { private void goDown(Health.Builder builder) {
builder.down(kafkaCommunicationResult.getException()).withDetail("topic", topic); builder.down(kafkaCommunicationResult.getException()).withDetail("topic", topic);
} }
} }

38
src/main/java/com/deviceinsight/kafka/health/KafkaHealthProperties.java
View File

@ -1,11 +1,13 @@
package com.deviceinsight.kafka.health; package com.deviceinsight.kafka.health;
import java.time.Duration;
public class KafkaHealthProperties { public class KafkaHealthProperties {
private String topic = "health-checks"; private String topic = "health-checks";
private long sendReceiveTimeoutMs = 2500; private Duration sendReceiveTimeout = Duration.ofMillis(2500);
private long pollTimeoutMs = 200; private Duration pollTimeout = Duration.ofMillis(200);
private long subscriptionTimeoutMs = 5000; private Duration subscriptionTimeout = Duration.ofSeconds(5);
public String getTopic() { public String getTopic() {
return topic; return topic;
@ -15,27 +17,33 @@ public class KafkaHealthProperties {
this.topic = topic; this.topic = topic;
} }
public long getSendReceiveTimeoutMs() { public Duration getSendReceiveTimeout() {
return sendReceiveTimeoutMs; return sendReceiveTimeout;
} }
public void setSendReceiveTimeoutMs(long sendReceiveTimeoutMs) { public void setSendReceiveTimeout(Duration sendReceiveTimeout) {
this.sendReceiveTimeoutMs = sendReceiveTimeoutMs; this.sendReceiveTimeout = sendReceiveTimeout;
} }
public long getPollTimeoutMs() { public Duration getPollTimeout() {
return pollTimeoutMs; return pollTimeout;
} }
public void setPollTimeoutMs(long pollTimeoutMs) { public void setPollTimeout(Duration pollTimeout) {
this.pollTimeoutMs = pollTimeoutMs; this.pollTimeout = pollTimeout;
} }
public long getSubscriptionTimeoutMs() { public Duration getSubscriptionTimeout() {
return subscriptionTimeoutMs; return subscriptionTimeout;
} }
public void setSubscriptionTimeoutMs(long subscriptionTimeoutMs) { public void setSubscriptionTimeout(Duration subscriptionTimeout) {
this.subscriptionTimeoutMs = subscriptionTimeoutMs; this.subscriptionTimeout = subscriptionTimeout;
}
@Override
public String toString() {
return "KafkaHealthProperties{" + "topic='" + topic + '\'' + ", sendReceiveTimeout=" + sendReceiveTimeout +
", pollTimeout=" + pollTimeout + ", subscriptionTimeout=" + subscriptionTimeout + '}';
} }
} }