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

GH-1138: HealthIndicator Improvements

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1138

Don't report DOWN if a container is stopped normally.
This is a valid state when containers are not auto-startup or are stopped while
the app remains running.

Containers are stopped abnormally when
- a listener throws an `Error`
- a `CommonContainerStoppingErrorHandler` (or similar) is configured to stop the
  container after an error.
This commit is contained in:
Gary Russell
2021-10-04 17:13:01 -04:00
parent 82a3306cb9
commit 2efd29fb27
3 changed files with 31 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -44,6 +44,7 @@
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>

4
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -201,10 +201,12 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
for (AbstractMessageListenerContainer<?, ?> container : listenerContainers) {
Map<String, Object> containerDetails = new HashMap<>();
boolean isRunning = container.isRunning();
if (!isRunning) {
boolean isOk = container.isInExpectedState();
if (!isOk) {
status = Status.DOWN;
}
containerDetails.put("isRunning", isRunning);
containerDetails.put("isStoppedAbnormally", !isRunning && !isOk);
containerDetails.put("isPaused", container.isContainerPaused());
containerDetails.put("listenerId", container.getListenerId());
containerDetails.put("groupId", container.getGroupId());

32
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
View File

@@ -108,8 +108,8 @@ public class KafkaBinderHealthIndicatorTest {
.willReturn(partitions);
org.mockito.BDDMockito.given(binder.getKafkaMessageListenerContainers())
.willReturn(Arrays.asList(listenerContainerA, listenerContainerB));
mockContainer(listenerContainerA, true);
mockContainer(listenerContainerB, true);
mockContainer(listenerContainerA, true, true);
mockContainer(listenerContainerB, true, true);
Health health = indicator.health();
assertThat(health.getStatus()).isEqualTo(Status.UP);
@@ -127,8 +127,27 @@ public class KafkaBinderHealthIndicatorTest {
.willReturn(partitions);
org.mockito.BDDMockito.given(binder.getKafkaMessageListenerContainers())
.willReturn(Arrays.asList(listenerContainerA, listenerContainerB));
mockContainer(listenerContainerA, false);
mockContainer(listenerContainerB, true);
mockContainer(listenerContainerA, false, true);
mockContainer(listenerContainerB, true, true);
Health health = indicator.health();
assertThat(health.getStatus()).isEqualTo(Status.UP);
assertThat(health.getDetails()).containsEntry("topicsInUse", singleton(TEST_TOPIC));
assertThat(health.getDetails()).hasEntrySatisfying("listenerContainers", value ->
assertThat((ArrayList<?>) value).hasSize(2));
}
@Test
public void kafkaBinderIsDownWhenOneOfContainersWasStoppedAbnormally() {
final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
"group1-healthIndicator", partitions, false));
org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
.willReturn(partitions);
org.mockito.BDDMockito.given(binder.getKafkaMessageListenerContainers())
.willReturn(Arrays.asList(listenerContainerA, listenerContainerB));
mockContainer(listenerContainerA, false, false);
mockContainer(listenerContainerB, true, true);
Health health = indicator.health();
assertThat(health.getStatus()).isEqualTo(Status.DOWN);
@@ -137,11 +156,14 @@ public class KafkaBinderHealthIndicatorTest {
assertThat((ArrayList<?>) value).hasSize(2));
}
private void mockContainer(AbstractMessageListenerContainer<?, ?> container, boolean isRunning) {
private void mockContainer(AbstractMessageListenerContainer<?, ?> container, boolean isRunning,
boolean normalState) {
org.mockito.BDDMockito.given(container.isRunning()).willReturn(isRunning);
org.mockito.BDDMockito.given(container.isContainerPaused()).willReturn(true);
org.mockito.BDDMockito.given(container.getListenerId()).willReturn("someListenerId");
org.mockito.BDDMockito.given(container.getGroupId()).willReturn("someGroupId");
org.mockito.BDDMockito.given(container.isInExpectedState()).willReturn(normalState);
}
@Test