Guozhang,
What does this mean if the changelog topic was disabled ? If thread 2 and
thread 4 are running in two different nodes and a rebalance occurs, thread 2
will not realize it is a zombie without the write to the changelog topic, right
? I am trying to understand the cases under which the changelog topic can ever
be disabled.
Thanks
Mohan
On 4/21/21, 10:22 PM, "Guozhang Wang" <wangg...@gmail.com> wrote:
Hello Mangat,
I think using persistent store that relies on in-memory stores could help
if the threads are from the same instance.
Guozhang
On Tue, Apr 20, 2021 at 12:54 AM mangat rai <mangatm...@gmail.com> wrote:
> Hey Guozhang,
>
> Thanks for creating the issue. Yes, you are right, this will happen only
> with the consecutive rebalancing as after some time zombie thread will
stop
> and re-join the group and the new thread will always overwrite the state
> with the latest data. In our poor infra setup, the rebalancing was
> happening many times in a row.
>
> Now, we can't guarantee that the consecutive rebalancing will not happen
> again (we reduced fetch-size which fixed it in many ways), will any of the
> following work as a workaround?
>
> 1. Use persistent store instead of in-memory. The new thread will never
get
> the lock hence we will lose availability but keep things consistent.
> 2. Use exactly-once semantics. However, we might need to redesign our
apps.
> It's a bigger change.
>
> Regards,
> Mangat
>
> On Tue, Apr 20, 2021 at 6:50 AM Guozhang Wang <wangg...@gmail.com> wrote:
>
> > Hello Mangat,
> >
> > What you've encountered is a "zombie writer" issue, that is, Thread-2
did
> > not know there's already a new rebalance and hence its partitions have
> been
> > migrated out, until it tries to commit and then got notified of the
> > illegal-generation error and realize itself is the "zombie" already.
This
> > case would still persist even with incremental rebalancing.
> >
> > I've filed https://issues.apache.org/jira/browse/KAFKA-12693 to
> summarize
> > the situation. Please LMK if that explanation is clear to you.
> >
> > On Mon, Apr 19, 2021 at 12:58 AM mangat rai <mangatm...@gmail.com>
> wrote:
> >
> > > Thanks, Guozhang,
> > >
> > > I was knocking myself with Kafka's various consumer rebalancing
> > algorithms
> > > in the last 2 days. Could I generalize this problem as
> > >
> > >
> > >
> > > *Any in-memory state store backed by a changelog topic will always
risk
> > > having interleaved writes from two different writers during
> rebalancing?*
> > > In our case, CPU throttling made it worse as thread-2 didn't try to
> > commit
> > > for a long time. Also,
> > >
> > > 1. Do you think if we disable the incremental rebalancing, we will not
> > have
> > > this issue because If I understood correctly Thread-4 will not start
> > > processing until the state is completely transferred from Thread-2.
> > > 2. If yes, how can we disable it without downgrading the client?
> > >
> > > Since we have a very low scale and no real-time computing requirement,
> we
> > > will be happy to sacrifice the availability to have consistency.
> > >
> > > Regards,
> > > Mangat
> > >
> > >
> > >
> > > On Sat, Apr 17, 2021 at 12:27 AM Guozhang Wang <wangg...@gmail.com>
> > wrote:
> > >
> > > > Hi Mangat:
> > > >
> > > > I think I found the issue of your problem here.
> > > >
> > > > It seems thread-2's partition was assigned to thread-4 while
thread-2
> > was
> > > > not aware (because it missed a rebalance, this is normal scenario);
> in
> > > > other words, thread2 becomes a "zombie". It would stay in that
zombie
> > > state
> > > > until it tried to commit, in which it would get an error from the
> > brokers
> > > > and realize its zombie identity and re-joins the group.
> > > >
> > > > During that period of time, before the commit was issued, it would
> > > continue
> > > > trying to write to its local states; here are several scenarios:
> > > >
> > > > 1) if thread-2/4 are belonging to two different nodes then that is
> > fine,
> > > > since they will write to different local state stores.
> > > > 2) if they belong to the same nodes, and
> > > > a) the state stores are persistent then they would have risks of
> > > > contention; this is guarded by the state directory locks (as file
> > locks)
> > > in
> > > > which case the new owner thread-4 should not be able to get on the
> > local
> > > > state files.
> > > > b) the state stores are in-memory, in which case that is fine
> since
> > > the
> > > > in-memory stores are kept separate as well.
> > > >
> > > > In your case: 2.b), the issue is that the changelog would still be
> > shared
> > > > between the two --- but note that this is the same case as in case
1)
> > as
> > > > well. And this means at that time the changelog is shared by two
> > writers
> > > > sending records interleaving. And if there’s a tombstone that was
> > > intended
> > > > for a record A, but when it was written interleaving and there’s
> > another
> > > > record B in between, that tombstone would effectively delete record
> B.
> > > The
> > > > key here is that, when we replay the changelogs, we replay it
> > completely
> > > > following offset ordering.
> > > >
> > > >
> > > >
> > > > On Thu, Apr 15, 2021 at 2:28 AM mangat rai <mangatm...@gmail.com>
> > wrote:
> > > >
> > > > > Guozhang,
> > > > >
> > > > > Yes, you are correct. We have our own group processor. I have more
> > > > > information now.
> > > > >
> > > > > 1. I added ThreadId in the data when the app persists into the
> > > changelog
> > > > > topic.
> > > > > 2. Thread-2 which was working with partition-0 had a timeout
issue.
> > > > > 4. Thread-4 picked up this partition-0 as I can see its Id in the
> > > > > changelog.
> > > > > 5. *But then Thread-2 and Thread-4 both were writing into the
> > > partition-0
> > > > > of the changelog, that too for the same key.*
> > > > >
> > > > > So I was clearly able to see that two threads were overwriting
data
> > of
> > > > one
> > > > > another into the state store leading to a corrupted state. This
> > > confirms
> > > > my
> > > > > theory that it was an issue of concurrent update. This was
> something
> > > > > totally unexpected. I suspect that Thread-2 continues to persist
> its
> > > > > in-memory state, maybe because It wasn't stopped after the timeout
> > > > > exception. Is there a configuration possible in the Kafka stream
> > which
> > > > > could lead to this?
> > > > >
> > > > > There was no network issue, our CPU was highly throttled by
> > Kubernetes.
> > > > We
> > > > > gave more resources, also decreased the fetch-size so we have more
> > I/O
> > > to
> > > > > Cpu time ratio than before, and then there was no timeout issue,
> > hence
> > > no
> > > > > reassignment and hence no corrupted state.
> > > > >
> > > > > I really appreciate your help here...
> > > > > Thanks!
> > > > > Mangat
> > > > >
> > > > >
> > > > > On Wed, Apr 14, 2021 at 8:48 PM Guozhang Wang <wangg...@gmail.com>
> > > > wrote:
> > > > >
> > > > > > Hey Mangat,
> > > > > >
> > > > > > A. With at least once, Streams does not make sure atomicity of
> 1) /
> > > 2);
> > > > > > with exactly once, atomicity is indeed guaranteed with
> > transactional
> > > > > > messaging.
> > > > > >
> > > > > > B. If you are using processor API, then I'm assuming you did
your
> > own
> > > > > > group-by processor right? In that case, the partition key would
> > just
> > > be
> > > > > the
> > > > > > record key when you are sending to the repartition topic.
> > > > > >
> > > > > >
> > > > > > Guozhang
> > > > > >
> > > > > >
> > > > > >
> > > > > >
> > > > > > On Thu, Apr 8, 2021 at 9:00 AM mangat rai <mangatm...@gmail.com>
> > > > wrote:
> > > > > >
> > > > > > > Thanks again, that makes things clear. I still have some
> > questions
> > > > here
> > > > > > > then.
> > > > > > >
> > > > > > > A. For each record we read, we do two updates
> > > > > > > 1. Changelog topic of the state store.
> > > > > > > 2. Output topic aka sink.
> > > > > > > Does the Kafka stream app make sure that either both are
> > > > > committed
> > > > > > or
> > > > > > > neither?
> > > > > > >
> > > > > > > B. Out Input topic actually has the as (a,b,c), but we
> partition
> > > > with
> > > > > > only
> > > > > > > (a). We do this because we have different compaction
> requirements
> > > > than
> > > > > > the
> > > > > > > partitions. It will still work as all (a,b,c) records will go
> to
> > > the
> > > > > same
> > > > > > > partition. Now in aggregation, we group by (a,b,c). In such
> case
> > > what
> > > > > > will
> > > > > > > be the partition key for the changelog topic?
> > > > > > >
> > > > > > > Note that we use low-level processor API and don't commit
> > > ourselves.
> > > > > > >
> > > > > > > Regards,
> > > > > > > Mangat
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > > On Thu, Apr 8, 2021 at 5:37 PM Guozhang Wang <
> wangg...@gmail.com
> > >
> > > > > wrote:
> > > > > > >
> > > > > > > > Hi Mangat,
> > > > > > > >
> > > > > > > > Please see my replies inline below.
> > > > > > > >
> > > > > > > > On Thu, Apr 8, 2021 at 5:34 AM mangat rai <
> > mangatm...@gmail.com>
> > > > > > wrote:
> > > > > > > >
> > > > > > > > > @Guozhang Wang
> > > > > > > > >
> > > > > > > > > Thanks for the reply. Indeed I am finding it difficult to
> > > > explain
> > > > > > this
> > > > > > > > > state. I checked the code many times. There can be a bug
> but
> > I
> > > > fail
> > > > > > to
> > > > > > > > see
> > > > > > > > > it. There are several things about the Kafka streams that
I
> > > don't
> > > > > > > > > understand, which makes it harder for me to reason.
> > > > > > > > >
> > > > > > > > > 1. What is the partition key for the changelog topics? Is
> it
> > > the
> > > > > same
> > > > > > > as
> > > > > > > > > the Input key or the state store key? Or maybe the thread
> > > > specifies
> > > > > > the
> > > > > > > > > partition as it knows the input partition it is subscribed
> > to?
> > > If
> > > > > the
> > > > > > > > input
> > > > > > > > > topic and state store are differently partitioned then we
> can
> > > > > explain
> > > > > > > the
> > > > > > > > > issue here.
> > > > > > > > >
> > > > > > > >
> > > > > > > > In Kafka Stream's changelog, the "partition key" of Kafka
> > > messages
> > > > is
> > > > > > the
> > > > > > > > same as the "message key" itself. And the message key is the
> > same
> > > > as
> > > > > > the
> > > > > > > > state store key.
> > > > > > > >
> > > > > > > > Since the state store here should be storing the running
> > > aggregate,
> > > > > it
> > > > > > > > means that the partition key is the same as the aggregated
> key.
> > > > > > > >
> > > > > > > > If you are doing a group-by aggregation here, where the
> > group-by
> > > > keys
> > > > > > are
> > > > > > > > different from the source input topic's keys, hence the
state
> > > store
> > > > > > keys
> > > > > > > > would be different with the input topic keys.
> > > > > > > >
> > > > > > > >
> > > > > > > > > 2. Is there a background thread to persist in the state
> store
> > > > when
> > > > > > > > caching
> > > > > > > > > is disabled? When will the app commit the log for the
input
> > > > topic?
> > > > > Is
> > > > > > > it
> > > > > > > > > when sink writes into the output topic or when the state
> > store
> > > > > writes
> > > > > > > > into
> > > > > > > > > the changelog topic? Because, if the app commits the
record
> > > > before
> > > > > > the
> > > > > > > > data
> > > > > > > > > was written to changelog topic then we can again explain
> this
> > > > state
> > > > > > > > >
> > > > > > > > > The commit happens *after* the local state store, as well
> as
> > > the
> > > > > > > > changelog records sent by the Streams' producers, have been
> > > > flushed.
> > > > > > I.e.
> > > > > > > > if there's a failure in between, you would re-process some
> > source
> > > > > > records
> > > > > > > > and hence cause duplicates, but no data loss (a.k.a. the
> > > > > at_least_once
> > > > > > > > semantics).
> > > > > > > >
> > > > > > > >
> > > > > > > >
> > > > > > > > > >You may also consider upgrading to 2.6.x or higher
version
> > and
> > > > see
> > > > > > if
> > > > > > > > this
> > > > > > > > > issue goes away.
> > > > > > > > > Do you mean the client or the Kafka broker? I will be
> > upgrading
> > > > the
> > > > > > > > client
> > > > > > > > > to 2.7.0 soon.
> > > > > > > > >
> > > > > > > > > I meant the client.
> > > > > > > >
> > > > > > > >
> > > > > > > > > Sadly looking into the timestamp will not help much as we
> use
> > > > some
> > > > > > > > business
> > > > > > > > > time field to set the record timestamp. If I am right,
> there
> > is
> > > > no
> > > > > > way
> > > > > > > > now
> > > > > > > > > to know that when a Producer wrote a record in a Kafka
> topic.
> > > > > > > > >
> > > > > > > > > Regards,
> > > > > > > > > Mangat
> > > > > > > > >
> > > > > > > > >
> > > > > > > > >
> > > > > > > > > On Wed, Apr 7, 2021 at 6:22 PM Guozhang Wang <
> > > wangg...@gmail.com
> > > > >
> > > > > > > wrote:
> > > > > > > > >
> > > > > > > > > > Hello Mangat,
> > > > > > > > > >
> > > > > > > > > > With at least once, although some records maybe
processed
> > > > > multiple
> > > > > > > > times
> > > > > > > > > > their process ordering should not be violated, so what
> you
> > > > > observed
> > > > > > > is
> > > > > > > > > not
> > > > > > > > > > expected. What caught my eyes are this section in your
> > output
> > > > > > > > changelogs
> > > > > > > > > > (high-lighted):
> > > > > > > > > >
> > > > > > > > > > Key1, V1
> > > > > > > > > > Key1, null
> > > > > > > > > > Key1, V1
> > > > > > > > > > Key1, null (processed again)
> > > > > > > > > > Key1, V2
> > > > > > > > > > Key1, null
> > > > > > > > > >
> > > > > > > > > > *Key1, V1Key1,V2*
> > > > > > > > > > Key1, V2+V1 (I guess we didn't process V2 tombstone yet
> but
> > > > > > > reprocessed
> > > > > > > > > V1
> > > > > > > > > > again due to reassignment)
> > > > > > > > > >
> > > > > > > > > > They seem to be the result of first receiving a
tombstone
> > > which
> > > > > > > removes
> > > > > > > > > V1
> > > > > > > > > > and then a new record that adds V2. However, since
> caching
> > is
> > > > > > > disabled
> > > > > > > > > you
> > > > > > > > > > should get
> > > > > > > > > >
> > > > > > > > > > *Key1,V1*
> > > > > > > > > > *Key1,null*
> > > > > > > > > > *Key1,V2*
> > > > > > > > > >
> > > > > > > > > > instead; without the actual code snippet I cannot tell
> more
> > > > > what's
> > > > > > > > > > happening here. If you can look into the logs you can
> > record
> > > > each
> > > > > > > time
> > > > > > > > > when
> > > > > > > > > > partition migrates, how many records from the changelog
> was
> > > > > > replayed
> > > > > > > to
> > > > > > > > > > restore the store, and from which offset on the input
> topic
> > > > does
> > > > > > > > Streams
> > > > > > > > > > resume processing. You may also consider upgrading to
> 2.6.x
> > > or
> > > > > > higher
> > > > > > > > > > version and see if this issue goes away.
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > Guozhang
> > > > > > > > > >
> > > > > > > > > > On Tue, Apr 6, 2021 at 8:38 AM mangat rai <
> > > > mangatm...@gmail.com>
> > > > > > > > wrote:
> > > > > > > > > >
> > > > > > > > > > > Hey,
> > > > > > > > > > >
> > > > > > > > > > > We have the following setup in our infrastructure.
> > > > > > > > > > >
> > > > > > > > > > > 1. Kafka - 2.5.1
> > > > > > > > > > > 2. Apps use kafka streams `org.apache.kafka`
version
> > > 2.5.1
> > > > > > > library
> > > > > > > > > > > 3. Low level processor API is used with
> *atleast-once*
> > > > > > semantics
> > > > > > > > > > > 4. State stores are *in-memory* with *caching
> > disabled*
> > > > and
> > > > > > > > > *changelog
> > > > > > > > > > > enabled*
> > > > > > > > > > >
> > > > > > > > > > >
> > > > > > > > > > > Is it possible that during state replication and
> > partition
> > > > > > > > > reassignment,
> > > > > > > > > > > the input data is not always applied to the state
> store?
> > > > > > > > > > >
> > > > > > > > > > > 1. Let's say the input topic is having records like
> > > following
> > > > > > > > > > >
> > > > > > > > > > > ```
> > > > > > > > > > > Key1, V1
> > > > > > > > > > > Key1, null (tombstone)
> > > > > > > > > > > Key1, V2
> > > > > > > > > > > Key1, null
> > > > > > > > > > > Key1, V3
> > > > > > > > > > > Key1, V4
> > > > > > > > > > > ```
> > > > > > > > > > > 2. The app has an aggregation function which takes
> these
> > > > record
> > > > > > and
> > > > > > > > > > update
> > > > > > > > > > > the state store so that changelog shall be
> > > > > > > > > > >
> > > > > > > > > > > ```
> > > > > > > > > > > Key1, V1
> > > > > > > > > > > Key1, null (tombstone)
> > > > > > > > > > > Key1, V2
> > > > > > > > > > > Key1, null
> > > > > > > > > > > Key1, V3
> > > > > > > > > > > Key1, V3 + V4
> > > > > > > > > > > ```
> > > > > > > > > > > Let's say the partition responsible for processing the
> > > above
> > > > > key
> > > > > > > was
> > > > > > > > > > > several times reallocated to different threads due to
> > some
> > > > > infra
> > > > > > > > issues
> > > > > > > > > > we
> > > > > > > > > > > are having(in Kubernetes where we run the app, not the
> > > Kafka
> > > > > > > > cluster).
> > > > > > > > > > >
> > > > > > > > > > > I see the following record in the changelogs
> > > > > > > > > > >
> > > > > > > > > > > ```
> > > > > > > > > > > Key1, V1
> > > > > > > > > > > Key1, null
> > > > > > > > > > > Key1, V1
> > > > > > > > > > > Key1, null (processed again)
> > > > > > > > > > > Key1, V2
> > > > > > > > > > > Key1, null
> > > > > > > > > > > Key1, V1
> > > > > > > > > > > Key1,V2
> > > > > > > > > > > Key1, V2+V1 (I guess we didn't process V2 tombstone
yet
> > but
> > > > > > > > reprocessed
> > > > > > > > > > V1
> > > > > > > > > > > again due to reassignment)
> > > > > > > > > > > Key1,V1 (V2 is gone as there was a tombstone, but then
> V1
> > > > > > tombstone
> > > > > > > > > > should
> > > > > > > > > > > have been applied also!!)
> > > > > > > > > > > Key1, V2+V1 (it is back!!!)
> > > > > > > > > > > Key1,V1
> > > > > > > > > > > Key1, V1 + V2 + V3 (This is the final state)!
> > > > > > > > > > > ```
> > > > > > > > > > >
> > > > > > > > > > > If you see this means several things
> > > > > > > > > > > 1. The state is always correctly applied locally (in
> > > > developer
> > > > > > > > laptop),
> > > > > > > > > > > where there were no reassignments.
> > > > > > > > > > > 2. The records are processed multiple times, which is
> > > > > > > understandable
> > > > > > > > as
> > > > > > > > > > we
> > > > > > > > > > > have at least symantics here.
> > > > > > > > > > > 3. As long as we re-apply the same events in the same
> > > orders
> > > > we
> > > > > > are
> > > > > > > > > > golden
> > > > > > > > > > > but looks like some records are skipped, but here it
> > looks
> > > as
> > > > > if
> > > > > > we
> > > > > > > > > have
> > > > > > > > > > > multiple consumers reading and update the same topics,
> > > > leading
> > > > > to
> > > > > > > > race
> > > > > > > > > > > conditions.
> > > > > > > > > > >
> > > > > > > > > > > Is there any way, Kafka streams' state replication
> could
> > > lead
> > > > > to
> > > > > > > > such a
> > > > > > > > > > > race condition?
> > > > > > > > > > >
> > > > > > > > > > > Regards,
> > > > > > > > > > > Mangat
> > > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > --
> > > > > > > > > > -- Guozhang
> > > > > > > > > >
> > > > > > > > >
> > > > > > > >
> > > > > > > >
> > > > > > > > --
> > > > > > > > -- Guozhang
> > > > > > > >
> > > > > > >
> > > > > >
> > > > > >
> > > > > > --
> > > > > > -- Guozhang
> > > > > >
> > > > >
> > > >
> > > >
> > > > --
> > > > -- Guozhang
> > > >
> > >
> >
> >
> > --
> > -- Guozhang
> >
>
--
-- Guozhang
