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

Merge branch 'develop'

Browse Source
This commit is contained in:
bvn13 2022-06-28 10:18:18 +03:00
commit 05005ec2d6
9 changed files with 133 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
README.adoc
View File

@ -1,115 +0,0 @@
= Kafka Health Check
:uri-build-status: https://travis-ci.org/deviceinsight/kafka-health-check
:img-build-status: https://api.travis-ci.org/deviceinsight/kafka-health-check.svg?branch=master
image:{img-build-status}[Build Status Badge,link={uri-build-status}]
This library provides a kafka health check for spring boot actuator.
== Usage
Add the following dependency to your `pom.xml`
[source,xml]
....
<dependency>
<groupId>com.deviceinsight.kafka</groupId>
<artifactId>kafka-health-check</artifactId>
<version>1.2.0</version>
</dependency>
....
In the same maven module you can configure the topic, poll timeouts, subscription timeouts and the receive timeouts
in the `application.yml`
An example for an `application.yaml` is:
[source,yaml]
....
kafka:
health:
topic: health-checks
sendReceiveTimeout: 2.5s
pollTimeout: 200ms
subscriptionTimeout: 5s
....
The values shown are the defaults.
IMPORTANT: Make sure the configured health check topic exists!
[source,java]
....
@Bean
@ConfigurationProperties("kafka.health")
public KafkaHealthProperties kafkaHealthProperties() {
return new KafkaHealthProperties();
}
....
[source,java]
....
@Bean
public KafkaConsumingHealthIndicator kafkaConsumingHealthIndicator(KafkaHealthProperties kafkaProperties,
KafkaProperties processingProperties) {
return new KafkaConsumingHealthIndicator(kafkaHealthProperties, processingProperties.buildConsumerProperties(),
processingProperties.buildProducerProperties());
}
....
Now if you call the actuator endpoint `actuator/health` you should see the following output:
[source,json]
....
{
"status" : "UP",
"details" : {
"kafkaConsuming" : {
"status" : "UP"
}
}
}
....
== Configuration
|===
|Property |Default |Description
|kafka.health.topic |`health-checks` | Topic to subscribe to
|kafka.health.sendReceiveTimeout |2.5s | The maximum time, given as https://docs.spring.io/spring-boot/docs/2.1.9.RELEASE/reference/html/boot-features-external-config.html#boot-features-external-config-conversion-duration[Duration], to wait for sending and receiving the message.
|kafka.health.pollTimeout |200ms | The time, given as https://docs.spring.io/spring-boot/docs/2.1.9.RELEASE/reference/html/boot-features-external-config.html#boot-features-external-config-conversion-duration[Duration], spent fetching the data from the topic
|kafka.health.subscriptionTimeout |5s | The maximum time, given as https://docs.spring.io/spring-boot/docs/2.1.9.RELEASE/reference/html/boot-features-external-config.html#boot-features-external-config-conversion-duration[Duration], to wait for subscribing to topic
|kafka.health.cache.maximumSize |200 | Specifies the maximum number of entries the cache may contain.
|===
== Releasing
Creating a new release involves the following steps:
. `./mvnw gitflow:release-start gitflow:release-finish`
. `git push origin master`
. `git push --tags`
. `git push origin develop`
In order to deploy the release to Maven Central, you need to create an account at https://issues.sonatype.org and
configure your account in `~/.m2/settings.xml`:
[source,xml]
....
<settings>
<servers>
<server>
<id>ossrh</id>
<username>your-jira-id</username>
<password>your-jira-pwd</password>
</server>
</servers>
</settings>
....
The account also needs access to the project on Maven Central. This can be requested by another project member.
Then check out the release you want to deploy (`git checkout x.y.z`) and run `./mvnw deploy -Prelease`.

104
README.md Normal file
View File

@ -0,0 +1,104 @@
# Kafka Health Check
> **Note.** _forked from [deviceinsight/kafka-health-check](https://github.com/deviceinsight/kafka-health-check) due to long period of inactivity_
This library provides a kafka health check for spring boot actuator.
## Usage
Add the following dependency to your `pom.xml`
```xml
<dependency>
<groupId>com.deviceinsight.kafka</groupId>
<artifactId>kafka-health-check</artifactId>
<version>1.4.0</version>
</dependency>
```
In the same maven module you can configure the topic, poll timeouts and the reception timeouts
in the `application.yaml`
An example for an `application.yaml` is:
```yaml
kafka:
health:
topic: health-checks
sendReceiveTimeout: 2.5s
pollTimeout: 200ms
```
The values shown are the defaults.
IMPORTANT: Make sure the configured health check topic exists!
```java
@Bean
@ConfigurationProperties("kafka.health")
public KafkaHealthProperties kafkaHealthProperties() {
return new KafkaHealthProperties();
}
```
```java
@Bean
public KafkaConsumingHealthIndicator kafkaConsumingHealthIndicator(KafkaHealthProperties kafkaProperties,
KafkaProperties processingProperties) {
return new KafkaConsumingHealthIndicator(kafkaHealthProperties, processingProperties.buildConsumerProperties(),
processingProperties.buildProducerProperties());
}
```
Now if you call the actuator endpoint `actuator/health` you should see the following output:
```json
{
"status" : "UP",
"details" : {
"kafkaConsuming" : {
"status" : "UP"
}
}
}
```
## Configuration
| Property | Default | Description |
|------------------------------------|------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| kafka.health.topic | `health-checks` | Topic to subscribe to |
| kafka.health.sendReceiveTimeout | 2.5s | The maximum time, given as [Duration](https://docs.spring.io/spring-boot/docs/2.1.9.RELEASE/reference/html/boot-features-external-config.html#boot-features-external-config-conversion-duration), to wait for sending and receiving the message. |
| kafka.health.pollTimeout | 200ms | The time, given as [Duration](https://docs.spring.io/spring-boot/docs/2.1.9.RELEASE/reference/html/boot-features-external-config.html#boot-features-external-config-conversion-duration), spent fetching the data from the topic |
| kafka.health.cache.maximumSize | 200 | Specifies the maximum number of entries the cache may contain. |
## Releasing
Creating a new release involves the following steps:
1. `./mvnw gitflow:release-start gitflow:release-finish`
2. `git push origin master`
3. `git push --tags`
4. `git push origin develop`
In order to deploy the release to Maven Central, you need to create an account at https://issues.sonatype.org and
configure your account in `~/.m2/settings.xml`:
```xml
<settings>
<servers>
<server>
<id>ossrh</id>
<username>your-jira-id</username>
<password>your-jira-pwd</password>
</server>
</servers>
</settings>
```
The account also needs access to the project on Maven Central. This can be requested by another project member.
Then check out the release you want to deploy (`git checkout x.y.z`) and run `./mvnw deploy -Prelease`.

26
pom.xml
View File

@ -4,19 +4,19 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.deviceinsight.kafka</groupId>
<groupId>me.bvn13.kafka.health</groupId>
<artifactId>kafka-health-check</artifactId>
<version>1.3.0</version>
<version>1.4.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Kafka Health Check</name>
<description>A kafka health check for spring boot actuator</description>
<url>https://github.com/deviceinsight/kafka-health-check</url>
<url>https://github.com/bvn13/kafka-health-check</url>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>2.4.4</version>
<version>2.7.1</version>
<relativePath />
</parent>
@ -133,6 +133,14 @@
</licenses>
<developers>
<developer>
<id>bvn13</id>
<name>Vyacheslav Boyko</name>
<email>dev@bvn13.me</email>
<roles>
<role>Developer</role>
</roles>
</developer>
<developer>
<id>ezienecker</id>
<name>Emanuel Zienecker</name>
@ -152,10 +160,10 @@
</developers>
<scm>
<connection>scm:git:git://github.com/deviceinsight/kafka-health-check.git</connection>
<developerConnection>scm:git:ssh://git@github.com:deviceinsight/kafka-health-check.git</developerConnection>
<connection>scm:git:git://github.com/bvn13/kafka-health-check.git</connection>
<developerConnection>scm:git:ssh://git@github.com:bvn13/kafka-health-check.git</developerConnection>
<tag>HEAD</tag>
<url>https://github.com/deviceinsight/kafka-health-check.git</url>
<url>https://github.com/bvn13/kafka-health-check.git</url>
</scm>
<profiles>
@ -195,7 +203,7 @@
<extensions>true</extensions>
<configuration>
<serverId>ossrh</serverId>
<nexusUrl>https://oss.sonatype.org/</nexusUrl>
<nexusUrl>https://s01.oss.sonatype.org/</nexusUrl>
<autoReleaseAfterClose>true</autoReleaseAfterClose>
</configuration>
</plugin>
@ -219,7 +227,7 @@
<distributionManagement>
<snapshotRepository>
<id>ossrh</id>
<url>https://oss.sonatype.org/content/repositories/snapshots</url>
<url>https://s01.oss.sonatype.org/content/repositories/snapshots</url>
</snapshotRepository>
</distributionManagement>

2
src/main/java/com/deviceinsight/kafka/health/KafkaCommunicationResult.java → src/main/java/me/bvn13/kafka/health/KafkaCommunicationResult.java
View File

@ -1,4 +1,4 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
final class KafkaCommunicationResult {

13
src/main/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicator.java → src/main/java/me/bvn13/kafka/health/KafkaConsumingHealthIndicator.java
View File

@ -1,4 +1,4 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@ -32,7 +32,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ -56,7 +55,6 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
private final String topic;
private final Duration sendReceiveTimeout;
private final Duration pollTimeout;
private final Duration subscriptionTimeout;
private final ExecutorService executor;
private final AtomicBoolean running;
@ -78,7 +76,6 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
this.topic = kafkaHealthProperties.getTopic();
this.sendReceiveTimeout = kafkaHealthProperties.getSendReceiveTimeout();
this.pollTimeout = kafkaHealthProperties.getPollTimeout();
this.subscriptionTimeout = kafkaHealthProperties.getSubscriptionTimeout();
Map<String, Object> kafkaConsumerPropertiesCopy = new HashMap<>(kafkaConsumerProperties);
@ -142,7 +139,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
private void subscribeToTopic() throws InterruptedException {
final CountDownLatch subscribed = new CountDownLatch(1);
final AtomicBoolean subscribed = new AtomicBoolean(false);
logger.info("Subscribe to health check topic={}", topic);
@ -150,7 +147,7 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// nothing to do her
// nothing to do here
}
@Override
@ -158,13 +155,13 @@ public class KafkaConsumingHealthIndicator extends AbstractHealthIndicator {
logger.debug("Got partitions = {}", partitions);
if (!partitions.isEmpty()) {
subscribed.countDown();
subscribed.set(true);
}
}
});
consumer.poll(pollTimeout);
if (!subscribed.await(subscriptionTimeout.toMillis(), MILLISECONDS)) {
if (!subscribed.get()) {
throw new BeanInitializationException("Subscription to kafka failed, topic=" + topic);
}

2
src/main/java/com/deviceinsight/kafka/health/KafkaHealthCheckCacheProperties.java → src/main/java/me/bvn13/kafka/health/KafkaHealthCheckCacheProperties.java
View File

@ -1,4 +1,4 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
public class KafkaHealthCheckCacheProperties {

18
src/main/java/com/deviceinsight/kafka/health/KafkaHealthProperties.java → src/main/java/me/bvn13/kafka/health/KafkaHealthProperties.java
View File

@ -1,4 +1,4 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
import java.time.Duration;
@ -7,7 +7,6 @@ public class KafkaHealthProperties {
private String topic = "health-checks";
private Duration sendReceiveTimeout = Duration.ofMillis(2500);
private Duration pollTimeout = Duration.ofMillis(200);
private Duration subscriptionTimeout = Duration.ofSeconds(5);
private KafkaHealthCheckCacheProperties cache = new KafkaHealthCheckCacheProperties();
public String getTopic() {
@ -44,19 +43,6 @@ public class KafkaHealthProperties {
setPollTimeout(Duration.ofMillis(pollTimeoutMs));
}
public Duration getSubscriptionTimeout() {
return subscriptionTimeout;
}
public void setSubscriptionTimeout(Duration subscriptionTimeout) {
this.subscriptionTimeout = subscriptionTimeout;
}
@Deprecated
public void setSubscriptionTimeoutMs(long subscriptionTimeoutMs) {
setSubscriptionTimeout(Duration.ofMillis(subscriptionTimeoutMs));
}
public KafkaHealthCheckCacheProperties getCache() {
return cache;
}
@ -68,7 +54,7 @@ public class KafkaHealthProperties {
@Override
public String toString() {
return "KafkaHealthProperties{" + "topic='" + topic + '\'' + ", sendReceiveTimeout=" + sendReceiveTimeout +
", pollTimeout=" + pollTimeout + ", subscriptionTimeout=" + subscriptionTimeout + ", cacheProperties=" +
", pollTimeout=" + pollTimeout + ", cacheProperties=" +
cache + '}';
}
}

5
src/test/java/com/deviceinsight/kafka/health/KafkaConsumingHealthIndicatorTest.java → src/test/java/me/bvn13/kafka/health/KafkaConsumingHealthIndicatorTest.java
View File

@ -1,9 +1,8 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
import static com.deviceinsight.kafka.health.KafkaConsumingHealthIndicatorTest.TOPIC;
import static me.bvn13.kafka.health.KafkaConsumingHealthIndicatorTest.TOPIC;
import static org.assertj.core.api.Assertions.assertThat;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import kafka.server.KafkaServer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.serialization.StringDeserializer;

5
src/test/java/com/deviceinsight/kafka/health/KafkaHealthPropertiesTest.java → src/test/java/me/bvn13/kafka/health/KafkaHealthPropertiesTest.java
View File

@ -1,4 +1,4 @@
package com.deviceinsight.kafka.health;
package me.bvn13.kafka.health;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.params.provider.Arguments.arguments;
@ -21,7 +21,6 @@ public class KafkaHealthPropertiesTest {
"kafka.health.topic", "custom-topic",
"kafka.health.send-receive-timeout", "1m",
"kafka.health.poll-timeout", "2s",
"kafka.health.subscription-timeout", "10s",
"kafka.health.cache.maximum-size", "42"
));
@ -29,7 +28,6 @@ public class KafkaHealthPropertiesTest {
"kafka.health.topic", "custom-topic",
"kafka.health.send-receive-timeout-ms", "60000",
"kafka.health.poll-timeout-ms", "2000",
"kafka.health.subscription-timeout-ms", "10000",
"kafka.health.cache.maximum-size", "42"
));
// @formatter:on
@ -46,7 +44,6 @@ public class KafkaHealthPropertiesTest {
assertThat(kafkaHealthProperties.getTopic()).isEqualTo("custom-topic");
assertThat(kafkaHealthProperties.getSendReceiveTimeout()).isEqualTo(Duration.ofMinutes(1));
assertThat(kafkaHealthProperties.getPollTimeout()).isEqualTo(Duration.ofSeconds(2));
assertThat(kafkaHealthProperties.getSubscriptionTimeout()).isEqualTo(Duration.ofSeconds(10));
assertThat(kafkaHealthProperties.getCache().getMaximumSize()).isEqualTo(42);
}