Hi Michael, When you call KafkaStreams.close(), it will first trigger a commitAll() function, which will 1) flush local state store if necessary; 2) flush messages buffered in producer; 3) commit offsets on consumer. Then it will close the producer / consumer clients and shutdown the tasks. So when you see processor's "close" function triggered, any buffered messages in the producer should already been flushed.
Did you see a different behavior than the above described? Guozhang On Fri, Apr 8, 2016 at 12:23 PM, Michael D. Coon <mdco...@yahoo.com.invalid> wrote: > All, > I'm seeing my processor's "close" method being called AFTER my > downstream producer has been closed. I had assumed that on close I would be > able to flush whatever I had been buffering up to send to kafka topic. In > other words, we've seen significant performance differences in building > flows with small messages and large messages in/out of kafka. So my > processor buffers up messages to a threshold and flushes those as a > composite message bundle to improve downstream processing. But if this > close method is called AFTER the producer has already been closed, I would > have no way to actually flush the final composite bundles to my topic on > shutdown. Is there some way to get a call BEFORE producer shutdown occurs? > Mike > > -- -- Guozhang
