kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Guozhang Wang <wangg...@gmail.com>
Subject Re: Kafka streams Processor life cycle behavior of close()
Date Tue, 04 Oct 2016 22:18:42 GMT
Created https://issues.apache.org/jira/browse/KAFKA-4253 for this issue.


Guozhang

On Tue, Oct 4, 2016 at 3:08 PM, Guozhang Wang <wangguoz@gmail.com> wrote:

> Hello Srikanth,
>
> We close the underlying clients before closing the state manager (hence
> the states) because for example we need to make sure producer's sent
> records have all been acked before the state manager records the changelog
> sent offsets as end offsets. This is kind of chicken-and-egg problem, and
> we may be able to re-order the shutting down process in the future with
> some added shutdown hooks.
>
> As of now, there is not a perfect solution to your scenario, and I would
> like to suggest checking if producer's own batching mechanism is good
> enough so you do not need to do this in the streams client layer.
>
>
> Guozhang
>
> On Sat, Oct 1, 2016 at 2:20 PM, Srikanth <srikanth.ht@gmail.com> wrote:
>
>> Hello,
>>
>> I'm testing out a WriteToSinkProcessor() that batches records before
>> writing it to a sink.
>> The actual commit to sink happens in punctuate(). I also wanted to commit
>> in close().
>> Idea here is, during a regular shutdown, we'll commit all records and
>> ideally stop with an empty state.
>> My commit() process is 3 step 1) Read from KV store 2) write to sink 3)
>> delete written keys from KV store.
>>
>> I get this exception when closing though. It looks like the kafka producer
>> is closed before the changelog topic is updated after close().
>> Should the producer be closed after all tasks and processors are closed?
>>
>> 16/10/01 17:01:15 INFO StreamThread-1 WriteToSinkProcessor: Closing
>> processor instance
>> 16/10/01 17:01:16 ERROR StreamThread-1 StreamThread: Failed to remove
>> stream tasks in thread [StreamThread-1]:
>> java.lang.IllegalStateException: Cannot send after the producer is
>> closed.
>>         at
>> org.apache.kafka.clients.producer.internals.RecordAccumulato
>> r.append(RecordAccumulator.java:173)
>>         at
>> org.apache.kafka.clients.producer.KafkaProducer.doSend(Kafka
>> Producer.java:467)
>>         at
>> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaPr
>> oducer.java:430)
>>         at
>> org.apache.kafka.streams.processor.internals.RecordCollector
>> .send(RecordCollector.java:84)
>>         at
>> org.apache.kafka.streams.processor.internals.RecordCollector
>> .send(RecordCollector.java:71)
>>         at
>> org.apache.kafka.streams.state.internals.StoreChangeLogger.
>> logChange(StoreChangeLogger.java:108)
>>         at
>> org.apache.kafka.streams.state.internals.InMemoryKeyValueLog
>> gedStore.flush(InMemoryKeyValueLoggedStore.java:161)
>>         at
>> org.apache.kafka.streams.state.internals.MeteredKeyValueStor
>> e.flush(MeteredKeyValueStore.java:165)
>>         at
>> org.apache.kafka.streams.processor.internals.ProcessorStateM
>> anager.close(ProcessorStateManager.java:343)
>>         at
>> org.apache.kafka.streams.processor.internals.AbstractTask.
>> close(AbstractTask.java:112)
>>         at
>> org.apache.kafka.streams.processor.internals.StreamTask.
>> close(StreamTask.java:317)
>>
>> Srikanth
>>
>
>
>
> --
> -- Guozhang
>



-- 
-- Guozhang

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message