I have the following Kafka stream setup
- Input topic has 6 partition
- Stream application has 6 threads.
- Single instance of the application is running at a moment.
- The topology has an in-memory state-store which is backed by a changelog topic.
- We are not using exactly-once semantics and have all other properties default.
- 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,
- Each thread in the topology will be allocated a partition each.
- State store is local to the thread and no other thread can see it.
- State store is also partitioned 6 == number of the threads.
- 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.
- 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?
- Does Kafka stop consuming the messages from thread-1 and then move the state?
- 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.
- The input topic has 0 lag for all partitions.
- change log reflects that the same input record was applied multiple times. This is understandable as we are using
at least oncesemantics. - 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?