Few questions about how Kafka Streams manages tasks

[Anish Mashankar <an...@systeminsights.com>]

First question: We know that Kafka Streams commits offsets on intervals.
But what offsets are committed? Are the offsets for messages committed are
the ones which have just arrived at the source node? Or the messages that
have been through the entire pipeline? If the latter, how do we avoid data
loss in this case? Is there an equivalent to Connect's RetriableException
in Kafka Streams?
----
Second question: Does giving an application host:port allows Kafka Streams
instances to communicate the data of state stores so that the entire
changelog topic is not read every time a crash happens?
----
Third question: If I want to upgrade the application using some kind of
code changes and bug fixes, how should the upgrade pattern proceed? Right
now, I just kill all the containers and bring new ones up. But this is
taking a lot of time as the topics are replayed. How can I do it faster?
Should I upgrade the processes slowly?
-- 

Regards,
Anish Samir Mashankar
R&D Engineer
System Insights
+91-9789870733

Re: Few questions about how Kafka Streams manages tasks

[Anish Mashankar <an...@systeminsights.com>]

Yes. I'm doing that. But in this case the container orchestration system is
killing it due to OOM or loss of node (I am simulating various failure
scenarios). But yeah, the fact that kafka streams only commits offsets of
messages that have been through the entire pipeline kind has enabled me to
take a state store out which has made recovery faster.

Thanks.

On Thu, Aug 17, 2017, 12:27 AM Guozhang Wang <wa...@gmail.com> wrote:

> I see.
>
> For normal maintenance operations, before you kill your container you could
> shuts down the Streams application by calling `KafkaStreams#close()`. Upon
> shutting down it would write a local checkpoint file indicating at which
> point in terms of offsets it has stopped at. So on resuming if the tasks
> gets allocated to the same host it can restart from that checkpoint offset
> position than wiping out the whole state and restarting from scratch.
>
> Guozhang
>
>
> On Tue, Aug 15, 2017 at 11:36 PM, Anish Mashankar <
> anish@systeminsights.com>
> wrote:
>
> > Hi Guozhang,
> > Thanks for the reply.
> > By taking a lot of time I meant that I see a log message `Restoring state
> > from changelog topics ....`, followed by just some kafka consumer logs
> like
> > `Discovered coordinator....`. Looking at this I assumed that the Stream
> > threads are waiting for the states to be restored. I am seeing this every
> > time an instance of the streams application fails and restarts.
> >
> > --
> > Anish Samir Mashankar
> > Site Reliability Engineer
> > VIMANA
> > +91-9789870733
> >
>
>
>
> --
> -- Guozhang
>
-- 

Regards,
Anish Samir Mashankar
R&D Engineer
System Insights
+91-9789870733

Re: Few questions about how Kafka Streams manages tasks

[Guozhang Wang <wa...@gmail.com>]

I see.

For normal maintenance operations, before you kill your container you could
shuts down the Streams application by calling `KafkaStreams#close()`. Upon
shutting down it would write a local checkpoint file indicating at which
point in terms of offsets it has stopped at. So on resuming if the tasks
gets allocated to the same host it can restart from that checkpoint offset
position than wiping out the whole state and restarting from scratch.

Guozhang


On Tue, Aug 15, 2017 at 11:36 PM, Anish Mashankar <an...@systeminsights.com>
wrote:

> Hi Guozhang,
> Thanks for the reply.
> By taking a lot of time I meant that I see a log message `Restoring state
> from changelog topics ....`, followed by just some kafka consumer logs like
> `Discovered coordinator....`. Looking at this I assumed that the Stream
> threads are waiting for the states to be restored. I am seeing this every
> time an instance of the streams application fails and restarts.
>
> --
> Anish Samir Mashankar
> Site Reliability Engineer
> VIMANA
> +91-9789870733
>



-- 
-- Guozhang

Re: Few questions about how Kafka Streams manages tasks

[Anish Mashankar <an...@systeminsights.com>]

Hi Guozhang,
Thanks for the reply.
By taking a lot of time I meant that I see a log message `Restoring state
from changelog topics ....`, followed by just some kafka consumer logs like
`Discovered coordinator....`. Looking at this I assumed that the Stream
threads are waiting for the states to be restored. I am seeing this every
time an instance of the streams application fails and restarts.

-- 
Anish Samir Mashankar
Site Reliability Engineer
VIMANA
+91-9789870733

Re: Few questions about how Kafka Streams manages tasks

[Guozhang Wang <wa...@gmail.com>]

Hi Anish,

1. The committed offsets are for the messages that have been through the
entire pipeline. Plus when we are committing, we make sure all state store
caches are flushed so there should be no messages that are "in the middle
of the topology". If there is a failure before the commit, then some
records may be re-fetched and re-processed, but no data-loss (i.e.
at-least-once). In 0.11.0 we add the exactly-once processing support to
avoid such duplicate message scenarios.

2. To avoid bootstrapping the state from the entire changelog, users can
set the number of standby-tasks to be non-zero. Standy-tasks simply drains
the changelog other active tasks are producing to, and build the local
state store images continuously. When previous active task crashes, the
library will try to migrate it to the hosts that have the corresponding
standby tasks if there are any. So that on these hosts the library do not
need to read from the beginning of the changelog to restore. Feel free to
read more details here:
https://kafka.apache.org/0110/documentation/streams/architecture#streams_architecture_recovery

3. "But this is taking a lot of time as the topics are replayed." I'm not
sure I understand this question. Do you mean that the apps never commit
offsets, so whenever they are resumed they always fetch from the beginning
of the source topics?


Guozhang



On Mon, Aug 14, 2017 at 9:55 PM, Anish Mashankar <an...@systeminsights.com>
wrote:

> First question: We know that Kafka Streams commits offsets on intervals.
> But what offsets are committed? Are the offsets for messages committed are
> the ones which have just arrived at the source node? Or the messages that
> have been through the entire pipeline? If the latter, how do we avoid data
> loss in this case? Is there an equivalent to Connect's RetriableException
> in Kafka Streams?
> ----
> Second question: Does giving an application host:port allows Kafka Streams
> instances to communicate the data of state stores so that the entire
> changelog topic is not read every time a crash happens?
> ----
> Third question: If I want to upgrade the application using some kind of
> code changes and bug fixes, how should the upgrade pattern proceed? Right
> now, I just kill all the containers and bring new ones up. But this is
> taking a lot of time as the topics are replayed. How can I do it faster?
> Should I upgrade the processes slowly?
> --
>
> Regards,
> Anish Samir Mashankar
> R&D Engineer
> System Insights
> +91-9789870733
>



-- 
-- Guozhang