Spring boot kafka streams doesn't terminate gracefully

Question

A very much default spring boot kafka application (https://github.com/spring-projects/spring-kafka/blob/master/src/reference/asciidoc/streams.adoc) does not terminate gracefully upon sigint.

The error being:

o.s.c.support.DefaultLifecycleProcessor  : Failed to shut down 1 bean with phase value 2147483547 within timeout of 30000: [org.springframework.kafka.config.internalKafkaListenerEndpointRegistry]

The skeleton of the application looks like:

@SpringBootApplication
public class ResponseAggregationsApplication {
    public static void main(String[] args) {
        SpringApplication.run(ResponseAggregationsApplication.class, args);
    }
}

Kafka Streams Configuration: This is not the full code, it's a simplistic version but I want to illustrate that I'm using the default annotations and such

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs(
            @Value("#{environment.KAFKA_BOOTSTRAP_SERVERS ?: 'kafka:9092'}") String bootstrapServers,
            @Value("#{environment.APPLICATION_ID ?: 'analysis-form-responses-aggregations'}") String applicationId,
            @Value("#{environment.KAFKA_NUM_STREAM_THREADS ?: 1}") int numThreads
    ) {
        this.feedbackResponsesTopic = feedbackResponsesTopic;
        this.formResponsesAggregationTopic = formResponsesAggregationTopic;
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public KStream<String, Item> kStream(
            StreamsBuilder kStreamBuilder
    ) {
        KStream<String, Item> stream = kStreamBuilder.stream(
                this.inputTopic,
                Consumed.with(Serdes.String(), this.itemSerde())
        );

        KGroupedStream<String, Item> groupedByAnotherId = stream
                .groupBy((key, item) -> item.getAnotherId());

        final KTable<String, Long> countAgg = groupedByAnotherId.count();

        countAgg.toStream().to(
                this.outputTopic, Produced.with(Serdes.String(), Serdes.String())
        );

        return stream;
    }
}

When I send the sigint (Via ctrl+c or stopping a docker container etc) the error is as follows:

2019-07-31 19:31:14.782 DEBUG 3041 --- [       Thread-8] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@4c9f7985, started on Wed Jul 31 19:31:06 CEST 2019
2019-07-31 19:31:14.782 DEBUG 3041 --- [       Thread-8] o.s.c.e.PropertySourcesPropertyResolver  : Found key 'spring.liveBeansView.mbeanDomain' in PropertySource 'systemProperties' with value of type String
2019-07-31 19:31:14.784 DEBUG 3041 --- [       Thread-8] o.s.c.support.DefaultLifecycleProcessor  : Stopping beans in phase 2147483547
2019-07-31 19:31:44.785  INFO 3041 --- [       Thread-8] o.s.c.support.DefaultLifecycleProcessor  : Failed to shut down 1 bean with phase value 2147483547 within timeout of 30000: [org.springframework.kafka.config.internalKafkaListenerEndpointRegistry]
2019-07-31 19:31:44.785 DEBUG 3041 --- [       Thread-8] o.s.c.support.DefaultLifecycleProcessor  : Stopping beans in phase 2147482647
2019-07-31 19:31:44.787  INFO 3041 --- [       Thread-8] org.apache.kafka.streams.KafkaStreams    : stream-client [analysis-form-responses-aggregations-702d79e4-f1a5-45d7-bb22-038d0c115409] State transition from RUNNING to PENDING_SHUTDOWN
2019-07-31 19:31:44.789  INFO 3041 --- [ms-close-thread] o.a.k.s.p.internals.StreamThread         : stream-thread [analysis-form-responses-aggregations-702d79e4-f1a5-45d7-bb22-038d0c115409-StreamThread-1] Informed to shut down
2019-07-31 19:31:44.789  INFO 3041 --- [ms-close-thread] o.a.k.s.p.internals.StreamThread         : stream-thread [analysis-form-responses-aggregations-702d79e4-f1a5-45d7-bb22-038d0c115409-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
2019-07-31 19:31:44.882  INFO 3041 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [analysis-form-responses-aggregations-702d79e4-f1a5-45d7-bb22-038d0c115409-StreamThread-1] Shutting down

So it seems that it fails to stop the bean org.springframework.kafka.config.internalKafkaListenerEndpointRegistry which is included via the KafkaBootstrapConfiguration, it says it comes from the @EnableKafka annotation, but I don't use it as you can see, I've tried with and without it but the effect is the same.

My versions are:

plugins {
    id 'org.springframework.boot' version '2.2.0.M4'
    id 'java'
}
dependencies {
    compile("org.springframework.boot:spring-boot-starter-actuator")
    compile 'io.sentry:sentry-logback:1.7.23'
    implementation 'org.springframework.kafka:spring-kafka:2.3.0.M3'
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.apache.kafka:kafka-streams'
    implementation 'org.apache.kafka:kafka-clients'
    implementation 'net.logstash.logback:logstash-logback-encoder:6.1'
    implementation 'ch.qos.logback:logback-core:1.2.3'
    implementation 'ch.qos.logback:logback-classic:1.2.3'
    testImplementation 'org.skyscreamer:jsonassert:1.5.0'
    testImplementation 'org.springframework.kafka:spring-kafka-test'
    testImplementation 'org.apache.kafka:kafka-streams-test-utils:2.3.0'
    testImplementation('org.springframework.boot:spring-boot-starter-test') {
        exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
        exclude group: 'junit', module: 'junit'
    }
    developmentOnly 'org.springframework.boot:spring-boot-devtools'
}

Answer 1

This is a known problem.

It's already fixed on all supported branches and will be in the next releases (due Monday).