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

Fixing Kafka streams binder health indicator tests

Browse Source 
Resolves #731
This commit is contained in:
Soby Chacko
2019-08-27 12:35:54 -04:00
committed by Oleg Zhurakousky
parent 8d5f794461
commit e53b0f0de9
2 changed files with 17 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,7 +34,7 @@ import org.springframework.boot.actuate.health.Status;
*
* @author Arnaud Jardiné
*/
class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
private final KafkaStreamsRegistry kafkaStreamsRegistry;

40
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java
View File

@@ -34,14 +34,15 @@ import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.actuate.health.CompositeHealthContributor;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.actuate.health.Status;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderHealthIndicator;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -114,23 +115,14 @@ public class KafkaStreamsBinderHealthIndicatorTests {
}
}
private static Status getStatusKStream(Map<String, Object> details) {
Health health = (Health) details.get("kstream");
return health != null ? health.getStatus() : Status.DOWN;
}
private static boolean waitFor(Map<String, Object> details) {
Health health = (Health) details.get("kstream");
if (health.getStatus() == Status.UP) {
Map<String, Object> moreDetails = health.getDetails();
Health kStreamHealth = (Health) moreDetails
.get("kafkaStreamsBinderHealthIndicator");
String status = (String) kStreamHealth.getDetails().get("threadState");
return status != null
&& (status.equalsIgnoreCase(KafkaStreams.State.REBALANCING.name())
|| status.equalsIgnoreCase("PARTITIONS_REVOKED")
|| status.equalsIgnoreCase("PARTITIONS_ASSIGNED")
|| status.equalsIgnoreCase(
private static boolean waitFor(Status status, Map<String, Object> details) {
if (status == Status.UP) {
String threadState = (String) details.get("threadState");
return threadState != null
&& (threadState.equalsIgnoreCase(KafkaStreams.State.REBALANCING.name())
|| threadState.equalsIgnoreCase("PARTITIONS_REVOKED")
|| threadState.equalsIgnoreCase("PARTITIONS_ASSIGNED")
|| threadState.equalsIgnoreCase(
KafkaStreams.State.PENDING_SHUTDOWN.name()));
}
return false;
@@ -183,15 +175,15 @@ public class KafkaStreamsBinderHealthIndicatorTests {
private static void checkHealth(ConfigurableApplicationContext context,
Status expected) throws InterruptedException {
HealthIndicator healthIndicator = context.getBean("bindersHealthIndicator",
HealthIndicator.class);
Health health = healthIndicator.health();
while (waitFor(health.getDetails())) {
CompositeHealthContributor healthIndicator = context
.getBean("bindersHealthContributor", CompositeHealthContributor.class);
KafkaStreamsBinderHealthIndicator kafkaStreamsBinderHealthIndicator = (KafkaStreamsBinderHealthIndicator) healthIndicator.getContributor("kstream");
Health health = kafkaStreamsBinderHealthIndicator.health();
while (waitFor(health.getStatus(), health.getDetails())) {
TimeUnit.SECONDS.sleep(2);
health = healthIndicator.health();
health = kafkaStreamsBinderHealthIndicator.health();
}
assertThat(health.getStatus()).isEqualTo(expected);
assertThat(getStatusKStream(health.getDetails())).isEqualTo(expected);
}
private ConfigurableApplicationContext singleStream() {