If I create and start a KafkaStream instance, and then in a shutdown hook call .close(), nothing happens - the logging indicates that the one StreamThread that exists has entered a PENDING_SHUTDOWN state, but then just sits like that forever. I've tried everything with no luck - have read the Kafka streams source code to see what it's doing during shutdown but to my eyes the code implies that the StreamThread, if in a running state, will never stop (which can't be right - I've not seen any bugs of this nature filed in the Kafka JIRA).
Here is the pertinent code of my simple KafkaStream (scala) application:
val props: Properties = {
val p = new Properties()
p.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-application")
p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
p.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "1")
p
}
implicit val produced = Produced.`with`(new StringSerde(), new StringSerde())
val builder: StreamsBuilder = new StreamsBuilder()
val in: KStream[String, String] = builder.stream[String, String]("input-topic")
in.map((k,v) =>{
println("Consumed and transforming value")
(k,s"$v_transformed")
}).to("output-topic")
val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
streams.start()
sys.addShutdownHook(streams.stop())
As you can see, it's just reading from one topic, making a tiny change to the value from the consumed record and writing it to another topic.
The stream is started and the application will call stop() when a SIGINT is sent to it.
When I ^C to terminate the process, I see Kafka logging stating that the StreamThread-1 is being transitioned to PENDING_SHUTDOWN and that's as far as it ever gets. It should eventually (within seconds) reach state NOT_RUNNING but it never does, and it continues to output my println statement for every record it's continuing to read from the input-topic.
What am I doing wrong here?
UPDATE: as per suggestion from commenter, i tried calling close() with a timeout of 60 seconds, and just got this eventually, but still no shutdown:-
[shutdownHook1] INFO o.apache.kafka.streams.KafkaStreams - stream-client [-] Streams client cannot stop completely within the timeout