Spring执行器+Kafka流-将Kafka流状态添加到健康检查endpoint

Question

management:
  health.db.enabled: false
  endpoints.web:
    base-path:
    path-mapping.health: /

当一个运行时异常被抛出并且我的流被停止时，日志显示，但是健康检查状态是向上的。

2019-09-17 13:16:31.522信息1---[Thread-5]org.apache.kafka.Streams.kafkastreams:streams-client[lpp-model-stream-7e6e8fea-fcad-4033-92a4-5ede50de6e17]Streams client停止completely

如何将kafka流状态绑定到健康检查endpoint？

  <dependencies>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-streams</artifactId>
            </dependency>
            <dependency>
                <groupId>data-wizards</groupId>
                <artifactId>lpp-common-avro</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-actuator</artifactId>
            </dependency>
            <dependency>
                <groupId>io.confluent</groupId>
                <artifactId>kafka-streams-avro-serde</artifactId>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>com.google.code.gson</groupId>
                <artifactId>gson</artifactId>
            </dependency>
            <dependency>
                <groupId>io.vavr</groupId>
                <artifactId>vavr</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>

Answer 1

KafkaStreams维护内存中的状态，该状态可以映射到执行器的健康状态。状态可以是以下状态之一:创建、错误、Not_running、pending_shutdown、重新平衡、running-它们是不言而喻的。有关状态转换，请参见文档https://kafka.apache.org/11/javadoc/org/apache/kafka/streams/kafkastreams.state.html

如果您正在寻找一个完整的示例，可以使用下面的示例并根据需要更新它（例如，您可能不将created算作状态向上）。确保在应用程序上下文中有kafkastreams类型的bean。

//Note that class name prefix before `HealthIndicator` will be camel-cased
//and used as a health component name, `kafkaStreams` here
@Component
public class KafkaStreamsHealthIndicator implements HealthIndicator {

    //if you have multiple instances, inject as Map<String, KafkaStreams>
    //Spring will map KafkaStreams instances by bean names present in context
    //so you can provide status details for each stream by name
    @Autowired
    private KafkaStreams kafkaStreams; 

    @Override
    public Health health() {
        State kafkaStreamsState = kafkaStreams.state();

        // CREATED, RUNNING or REBALANCING
        if (kafkaStreamsState == State.CREATED || kafkaStreamsState.isRunning()) {
            //set details if you need one
            return Health.up().build();
        }

        // ERROR, NOT_RUNNING, PENDING_SHUTDOWN, 
        return Health.down().withDetail("state", kafkaStreamsState.name()).build();
    }
}

然后健康endpoint将显示如下：

{
    "status": "UP",
    "kafkaStreams": {
        "status": "DOWN",
        "details": {  //not included if "UP"
            "state": "NOT_RUNNING"
        }
    }
}