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

Merge pull request #4 from deviceinsight/feature/refactor-health-strategy

Browse Source 
Refactor health check strategy
This commit is contained in:
Manu 2019-06-03 15:50:32 +02:00 committed by GitHub
commit a0c5c925c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 105 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
changelog.adoc
View File

@ -4,7 +4,8 @@
== 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.
* Refactor health check strategy: Kafka polled continuously.
== Version 0.1.0 == Version 1.0.0
* Develop kafka health check * Develop kafka health check

40
pom.xml
View File

@ -23,19 +23,19 @@
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<!-- Versions --> <!-- Versions -->
<junit.jupiter.version>5.3.2</junit.jupiter.version> <spring-boot.version>2.1.5.RELEASE</spring-boot.version>
<docker-maven-plugin.version>0.27.2</docker-maven-plugin.version>
<gitflow-maven-plugin.version>1.12.0</gitflow-maven-plugin.version>
<spring-boot.version>2.1.3.RELEASE</spring-boot.version>
<spring.kafka.version>2.2.4.RELEASE</spring.kafka.version> <spring.kafka.version>2.2.4.RELEASE</spring.kafka.version>
<caffeine.version>2.7.0</caffeine.version>
<awaitility.version>3.1.6</awaitility.version>
<junit.jupiter.version>5.4.2</junit.jupiter.version>
<assertj-core.version>3.11.1</assertj-core.version>
<maven-source-plugin.version>3.0.1</maven-source-plugin.version> <maven-source-plugin.version>3.0.1</maven-source-plugin.version>
<maven-surefire-plugin.version>2.22.1</maven-surefire-plugin.version> <maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
<assertj-core.version>3.11.1</assertj-core.version>
<guava.version>27.1-jre</guava.version>
<nexus-staging-maven-plugin.version>1.6.8</nexus-staging-maven-plugin.version> <nexus-staging-maven-plugin.version>1.6.8</nexus-staging-maven-plugin.version>
<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>
</properties> </properties>
<dependencies> <dependencies>
@ -52,26 +52,14 @@
<scope>provided</scope> <scope>provided</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.guava</groupId> <groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>guava</artifactId> <artifactId>caffeine</artifactId>
<version>${guava.version}</version> <version>${caffeine.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.junit.jupiter</groupId> <groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId> <artifactId>junit-jupiter</artifactId>
<version>${junit.jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-params</artifactId>
<version>${junit.jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>${junit.jupiter.version}</version> <version>${junit.jupiter.version}</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
@ -93,6 +81,12 @@
<version>${spring.kafka.version}</version> <version>${spring.kafka.version}</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>${awaitility.version}</version>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
<build> <build>

22
src/main/java/com/deviceinsight/kafka/health/KafkaCommunicationResult.java
View File

@ -2,30 +2,22 @@ package com.deviceinsight.kafka.health;
final class KafkaCommunicationResult { final class KafkaCommunicationResult {
private final String topic;
private final Exception exception; private final Exception exception;
private KafkaCommunicationResult(String topic) { private KafkaCommunicationResult() {
this.topic = topic;
this.exception = null; this.exception = null;
} }
private KafkaCommunicationResult(String topic, Exception exception) { private KafkaCommunicationResult(Exception exception) {
this.topic = topic;
this.exception = exception; this.exception = exception;
} }
static KafkaCommunicationResult success(String topic) { static KafkaCommunicationResult success() {
return new KafkaCommunicationResult(topic); return new KafkaCommunicationResult();
} }
static KafkaCommunicationResult failure(String topic, Exception exception) { static KafkaCommunicationResult failure(Exception exception) {
return new KafkaCommunicationResult(topic, exception); return new KafkaCommunicationResult(exception);
}
String getTopic() {
return topic;
} }
Exception getException() { Exception getException() {
@ -34,7 +26,7 @@ final class KafkaCommunicationResult {
@Override @Override
public String toString() { public String toString() {
return "KafkaCommunication{topic='" + topic + "', exception=" + exception + '}'; return "KafkaCommunication{exception=" + exception + '}';
} }
public boolean isFailure() { public boolean isFailure() {

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

@ -2,10 +2,12 @@ package com.deviceinsight.kafka.health;
import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.MILLISECONDS;
import com.google.common.annotations.VisibleForTesting; import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
@ -15,6 +17,7 @@ import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.boot.actuate.health.AbstractHealthIndicator; import org.springframework.boot.actuate.health.AbstractHealthIndicator;
import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.Health;
@ -29,12 +32,11 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future; import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeoutException;
import java.util.stream.StreamSupport; import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
@ -54,7 +56,10 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
private final long subscriptionTimeoutMs; private final long subscriptionTimeoutMs;
private final ExecutorService executor; private final ExecutorService executor;
private final AtomicBoolean running;
private final Cache<String, String> cache;
private KafkaCommunicationResult kafkaCommunicationResult;
public KafkaConsumingHealthIndicator(KafkaHealthProperties kafkaHealthProperties, public KafkaConsumingHealthIndicator(KafkaHealthProperties kafkaHealthProperties,
Map<String, Object> kafkaConsumerProperties, Map<String, Object> kafkaProducerProperties) { Map<String, Object> kafkaConsumerProperties, Map<String, Object> kafkaProducerProperties) {
@ -74,22 +79,34 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
this.consumer = new KafkaConsumer<>(kafkaConsumerPropertiesCopy, deserializer, deserializer); this.consumer = new KafkaConsumer<>(kafkaConsumerPropertiesCopy, deserializer, deserializer);
this.producer = new KafkaProducer<>(kafkaProducerProperties, serializer, serializer); this.producer = new KafkaProducer<>(kafkaProducerProperties, serializer, serializer);
this.executor = new ThreadPoolExecutor(0, 1, 0L, MILLISECONDS, new SynchronousQueue<>(), this.executor = Executors.newSingleThreadExecutor();
new ThreadPoolExecutor.AbortPolicy()); this.running = new AtomicBoolean(true);
this.cache = Caffeine.newBuilder().expireAfterWrite(sendReceiveTimeoutMs, TimeUnit.MILLISECONDS).build();
this.kafkaCommunicationResult =
KafkaCommunicationResult.failure(new RejectedExecutionException("Kafka Health Check is starting."));
} }
@PostConstruct @PostConstruct
void subscribeAndSendMessage() throws InterruptedException { void subscribeAndSendMessage() throws InterruptedException {
subscribeToTopic(); subscribeToTopic();
KafkaCommunicationResult kafkaCommunicationResult = sendAndReceiveMessage();
if (kafkaCommunicationResult.isFailure()) { if (kafkaCommunicationResult.isFailure()) {
throw new RuntimeException("Kafka health check failed", kafkaCommunicationResult.getException()); throw new BeanInitializationException("Kafka health check failed", kafkaCommunicationResult.getException());
} }
executor.submit(() -> {
while (running.get()) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
records.forEach(record -> cache.put(record.key(), record.value()));
}
});
} }
@PreDestroy @PreDestroy
void shutdown() { void shutdown() {
executor.shutdown(); running.set(false);
executor.shutdownNow();
producer.close(); producer.close();
consumer.close(); consumer.close();
} }
@ -105,8 +122,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
} }
} }
@VisibleForTesting private void subscribeToTopic() throws InterruptedException {
void subscribeToTopic() throws InterruptedException {
final CountDownLatch subscribed = new CountDownLatch(1); final CountDownLatch subscribed = new CountDownLatch(1);
@ -131,68 +147,75 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
consumer.poll(Duration.ofMillis(pollTimeoutMs)); consumer.poll(Duration.ofMillis(pollTimeoutMs));
if (!subscribed.await(subscriptionTimeoutMs, MILLISECONDS)) { if (!subscribed.await(subscriptionTimeoutMs, MILLISECONDS)) {
throw new RuntimeException("Subscription to kafka failed, topic=" + topic); throw new BeanInitializationException("Subscription to kafka failed, topic=" + topic);
} }
this.kafkaCommunicationResult = KafkaCommunicationResult.success();
} }
private KafkaCommunicationResult sendAndReceiveMessage() { private String sendMessage() {
Future<Void> sendReceiveTask = null;
try { try {
return sendKafkaMessage();
sendReceiveTask = executor.submit(() -> {
sendAndReceiveKafkaMessage();
return null;
});
sendReceiveTask.get(sendReceiveTimeoutMs, MILLISECONDS);
} catch (ExecutionException e) { } catch (ExecutionException e) {
logger.warn("Kafka health check execution failed.", e); logger.warn("Kafka health check execution failed.", e);
return KafkaCommunicationResult.failure(topic, e); this.kafkaCommunicationResult = KafkaCommunicationResult.failure(e);
} catch (TimeoutException | InterruptedException e) { } catch (TimeoutException | InterruptedException e) {
logger.warn("Kafka health check timed out.", e); logger.warn("Kafka health check timed out.", e);
sendReceiveTask.cancel(true); this.kafkaCommunicationResult = KafkaCommunicationResult.failure(e);
return KafkaCommunicationResult.failure(topic, e);
} catch (RejectedExecutionException e) { } catch (RejectedExecutionException e) {
logger.debug("Ignore health check, already running..."); logger.debug("Ignore health check, already running...");
} }
return KafkaCommunicationResult.success(topic);
return null;
} }
private void sendAndReceiveKafkaMessage() throws Exception { private String sendKafkaMessage() throws InterruptedException, ExecutionException, TimeoutException {
String message = UUID.randomUUID().toString(); String message = UUID.randomUUID().toString();
long startTime = System.currentTimeMillis();
logger.debug("Send health check message = {}", message); logger.debug("Send health check message = {}", message);
producer.send(new ProducerRecord<>(topic, message, message)).get(sendReceiveTimeoutMs, MILLISECONDS); producer.send(new ProducerRecord<>(topic, message, message)).get(sendReceiveTimeoutMs, MILLISECONDS);
while (messageNotReceived(message)) { return message;
logger.debug("Waiting for message={}", message);
}
logger.debug("Kafka health check succeeded. took= {} msec", System.currentTimeMillis() - startTime);
} }
private boolean messageNotReceived(String message) {
return StreamSupport.stream(consumer.poll(Duration.ofMillis(pollTimeoutMs)).spliterator(), false)
.noneMatch(msg -> msg.key().equals(message) && msg.value().equals(message));
}
@Override @Override
protected void doHealthCheck(Health.Builder builder) { protected void doHealthCheck(Health.Builder builder) {
KafkaCommunicationResult kafkaCommunicationResult = sendAndReceiveMessage(); String expectedMessage = sendMessage();
if (expectedMessage == null) {
if (kafkaCommunicationResult.isFailure()) { goDown(builder);
builder.down(kafkaCommunicationResult.getException()) return;
.withDetail("topic", kafkaCommunicationResult.getTopic());
} else {
builder.up();
} }
long startTime = System.currentTimeMillis();
while (true) {
String receivedMessage = cache.getIfPresent(expectedMessage);
if (expectedMessage.equals(receivedMessage)) {
builder.up();
return;
} else if (System.currentTimeMillis() - startTime > sendReceiveTimeoutMs) {
if (kafkaCommunicationResult.isFailure()) {
goDown(builder);
} else {
builder.down(new TimeoutException(
"Sending and receiving took longer than " + sendReceiveTimeoutMs + " ms"))
.withDetail("topic", topic);
}
return;
}
}
} }
private void goDown(Health.Builder builder) {
builder.down(kafkaCommunicationResult.getException()).withDetail("topic", topic);
}
} }

12
src/test/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicatorTest.java
View File

@ -6,7 +6,10 @@ import static org.assertj.core.api.Assertions.assertThat;
import kafka.server.KafkaServer; import kafka.server.KafkaServer;
import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.*; import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.Health;
@ -25,7 +28,7 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
@ExtendWith(SpringExtension.class) @ExtendWith(SpringExtension.class)
@EmbeddedKafka(topics = {TOPIC}) @EmbeddedKafka(topics = TOPIC)
public class KafkaConsumingHealthIndicatorTest { public class KafkaConsumingHealthIndicatorTest {
static final String TOPIC = "health-checks"; static final String TOPIC = "health-checks";
@ -64,15 +67,14 @@ public class KafkaConsumingHealthIndicatorTest {
final KafkaConsumingHealthIndicator healthIndicator = final KafkaConsumingHealthIndicator healthIndicator =
new KafkaConsumingHealthIndicator(kafkaHealthProperties, kafkaProperties.buildConsumerProperties(), new KafkaConsumingHealthIndicator(kafkaHealthProperties, kafkaProperties.buildConsumerProperties(),
kafkaProperties.buildProducerProperties()); kafkaProperties.buildProducerProperties());
healthIndicator.subscribeToTopic(); healthIndicator.subscribeAndSendMessage();
Health health = healthIndicator.health(); Health health = healthIndicator.health();
assertThat(health.getStatus()).isEqualTo(Status.UP); assertThat(health.getStatus()).isEqualTo(Status.UP);
shutdownKafka(); shutdownKafka();
health = healthIndicator.health(); Awaitility.await().untilAsserted(() -> assertThat(healthIndicator.health().getStatus()).isEqualTo(Status.DOWN));
assertThat(health.getStatus()).isEqualTo(Status.DOWN);
} }
private void shutdownKafka() { private void shutdownKafka() {