kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Srikanth <srikanth...@gmail.com>
Subject Re: Kafka streams Processor life cycle behavior of close()
Date Sun, 09 Oct 2016 00:00:44 GMT
Tnx! Looks like fix is already in for 0.10.1.0

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

> 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