Kafka Stream: How does rebalancing works between threads?

Viewed 649

I have the following Kafka stream setup

  1. Input topic has 6 partition
  2. Stream application has 6 threads.
  3. Single instance of the application is running at a moment.
  4. The topology has an in-memory state-store which is backed by a changelog topic.
  5. We are not using exactly-once semantics and have all other properties default.
  6. Data in the input topic is partitioned with a key. The partitioning is reproducible. I.e. same keys will always give the same partition.

If I understand right, in such a setup,

  1. Each thread in the topology will be allocated a partition each.
  2. State store is local to the thread and no other thread can see it.
  3. State store is also partitioned 6 == number of the threads.
  4. I confirmed the above by running the application locally.

So far so good, but our production lately has various network interruptions leading to brokers being unable to talk to each other. Here the things get a little tricky and hard to understand. We got the following alerts within 5 seconds of that happening

[Consumer clientId=6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-1-consumer, groupId=v11] 
Offset commit failed on partition topic.v5-0 at offset 180855: The coordinator is not aware of this member.
severity: ERROR
thread: 6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-1
logger: org.apache.kafka.clients.consumer.internals.ConsumerCoordinator
Show more



[Consumer clientId=6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-2-consumer, groupId=v11] 
Offset commit failed on partition topic.v5-0 at offset 189332: The coordinator is not aware of this member.
thread: 6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-2
logger: org.apache.kafka.clients.consumer.internals.ConsumerCoordinator
Show more


[Consumer clientId=6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-6-consumer, groupId=v11] 
Offset commit failed on partition topic.v5-0 at offset 340410: The coordinator is not aware of this member.
thread: 6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-6
logger: org.apache.kafka.clients.consumer.internals.ConsumerCoordinator
Show more


[Consumer clientId=6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-1-consumer, groupId=v11] 
Offset commit failed on partition topic.v5-0 at offset 344410: The coordinator is not aware of this member.

thread: 6b4fcfec-c94a-4a35-b37a-f7755f45415d-StreamThread-1
logger: org.apache.kafka.clients.consumer.internals.ConsumerCoordinator
Show more

From here it looks like either the same partition is read by multiple threads, or Kafka marks a consumer in a thread as FAILED and then moves the processing to the next consumer an so on. Here are several things I don't understand.

  1. Stream thread-2 is already subscribed to another partition, what happens to its state store? Does the state store from the Stream thread-1 merged into the 2nd one?
  2. Does Kafka stop consuming the messages from thread-1 and then move the state?
  3. Does stream-1 starts processing its partition again after a while, or the processing for this partition is forever moved?

The reason I am asking this is that I see the following behavior when I look into the changelog topic.

  1. The input topic has 0 lag for all partitions.
  2. change log reflects that the same input record was applied multiple times. This is understandable as we are using at least once semantics.
  3. However the final state store is inaccurate. The inaccuracy can only be explained if either two stream-threads were updating the same changelog partitions, or the state from the state-store was not flushed completely into the changelog topic

Could anyone please help me to understand how does Kafka rebalances the stream applications in such a scenario and if there is any known bug related to the same?

1 Answers

The Streams rebalance protocol is quite complicated, there are a couple blog posts that explain it better than I can: rebalance protocol and static membership.

A quick answer as to why a stream thread might have failed is because when a thread can not talk to the group coordinator long enough, for any reason, it get kicked out of the group and its tasks get reassigned. If it is just a network issue the thread can recover and be reassigned tasks later after rejoining the group.

From reading your question it seems that you really want to know why the final state store is inaccurate. State is not guaranteed to be uncorrupted if there are errors such as as network partitions in at least once, a thread might not realize it is out of a group and keep processing. For that you need EOS. EOS_beta does have better performance in more recent releases.

Related