KafkaBinderHealthIndicator implements HealthIndicator, DisposableBean { @Override public Health health() { Future<Health> future = executor.submit(this::buildHealthStatus); try { return future.get(this.timeout, TimeUnit.SECONDS); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); return Health.down() .withDetail("Interrupted while waiting for partition information in", this.timeout + " seconds") .build(); } catch (ExecutionException ex) { return Health.down(ex).build(); } catch (TimeoutException ex) { return Health.down().withDetail("Failed to retrieve partition information in", this.timeout + " seconds").build(); } } KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
			ConsumerFactory<?, ?> consumerFactory); void setTimeout(int timeout); void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader); @Override Health health(); @Override void destroy();  }
@Test public void createsConsumerOnceWhenInvokedMultipleTimes() { final List<PartitionInfo> partitions = partitions(new Node(0, null, 0)); topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( "group4-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) .willReturn(partitions); indicator.health(); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UP); org.mockito.Mockito.verify(this.consumerFactory).createConsumer(); }
@Test public void consumerCreationFailsFirstTime() { final List<PartitionInfo> partitions = partitions(new Node(0, null, 0)); topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( "foo-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumerFactory.createConsumer()) .willThrow(KafkaException.class).willReturn(consumer); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.DOWN); health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UP); org.mockito.Mockito.verify(this.consumerFactory, Mockito.times(2)) .createConsumer(); }
@Test public void testIfNoTopicsRegisteredByTheBinderProvidesDownStatus() { Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UNKNOWN); }
@Test public void kafkaBinderIsUp() { 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); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UP); }
@Test public void kafkaBinderIsUpWithRegexTopic() { topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation( "regex-healthIndicator", null, true)); Health health = indicator.health(); org.mockito.BDDMockito.verify(consumer, Mockito.never()) .partitionsFor(REGEX_TOPIC); assertThat(health.getStatus()).isEqualTo(Status.UP); }
@Test public void kafkaBinderIsDown() { final List<PartitionInfo> partitions = partitions(null); topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( "group2-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) .willReturn(partitions); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.DOWN); }