[
https://issues.apache.org/jira/browse/KAFKA-15035?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sam Cantero updated KAFKA-15035:
--------------------------------
Description:
We've recently encountered a scenario where a consumer group got their
committed offsets deleted some minutes (around 3 minutes) after the consumer
got into inactive state (the underlying node went away).
As per
[KIP-211|https://cwiki.apache.org/confluence/display/KAFKA/KIP-211%3A+Revise+Expiration+Semantics+of+Consumer+Group+Offsets],
committed offsets for an active (i.e running) consumer group should not be
deleted. However, if a consumer becomes inactive, {+}the deletion of committed
offsets will not occur immediately{+}. Instead, the committed offsets will only
be removed if the consumer remains inactive for at least the duration specified
by
[offset.retention.minutes|https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes].
In our case {{offset.retention.minutes}} is set to 7 days and the consumer was
only inactive for 5 minutes, so deletion should have not occurred.
Later on KIP-496 was introduced to fix the following issue in KIP-211:
{quote}When a consumer subscription changes, we are still left with the
committed offsets of the previous subscription. These will never be cleaned up
as long as the group remains active. We were aware of this problem in KIP-211,
but the solution was not implemented because the coordinator is presently
agnostic to join group metadata and we were unclear about the compatibility
implications of changing that.
{quote}
However this introduced a regression as explained in
https://issues.apache.org/jira/browse/KAFKA-13636.
{quote}The group coordinator might delete invalid offsets during a group
rebalance. During a rebalance, the coordinator is relying on the last commit
timestamp ({_}offsetAndMetadata.commitTimestamp{_}) instead of the last state
modification {_}timestamp (currentStateTimestamp{_}) to detect expired offsets.
{quote}
It is implied, that Kafka employs two approaches for offset expiration:
* The deletion timer is activated when a consumer group enters the Empty state
(i.e., not running). Once the timer exceeds the {{offset.retention.minutes}}
threshold, the committed offsets are deleted.
* If a consumer is in a "running" state (i.e., not in the Empty state) but is
no longer consuming from topics with committed offsets older than the
offset.retention.minutes duration, the committed offsets are deleted.
However, the Kafka issue KAFKA-13636 specifically states that this situation
could occur during a group rebalance. In my particular scenario, I have
observed that the affected consumer group did not transition into the Empty
state, and I have encountered a "Stabilized" log line. I'm uncertain whether
this log line indicates that the rebalance had concluded at that point or not.
The reason why Kafka did not detect this consumer group as Empty remains
unclear.
*Logs*
{noformat}
01:30:47 am - [GroupCoordinator 1]: Member consumer-mycg-1-uuid in group mycg
has failed, removing it from the group
01:30:47 am - [GroupCoordinator 1]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 432 (__consumer_offsets-16) (reason:
removing member consumer-mycg-1-uuid on heartbeat expiration)
1:30:50 am - [GroupCoordinator 1]: Member consumer-mycg-2-uuid in group mycg
has failed, removing it from the group
01:30:50 am - [GroupCoordinator 1]: Stabilized group mycg generation 433
(__consumer_offsets-16)
01:30:50 am - [GroupCoordinator 1]: Assignment received from leader for group
mycg for generation 433{noformat}
This suggests that kafka might have followed the second approach and that's why
kafka deleted the offsets 3 minutes later.
{noformat}
1:33:17 am -
[GroupMetadataManager brokerId=1] Removed 285 expired offsets in 8
milliseconds.{noformat}
As a reference a regular consumer join/startup logs looks like this. The group
is stabilised and the assignment from the leader received.
{noformat}
[GroupCoordinator 0]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 6 (__consumer_offsets-22) (reason:
Adding new member consumer-mycg-1-2b8ba689-fbaa-4829-82f5-dd2ed1d89d86 with
group instance id None) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Stabilized group mycg generation 7
(__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Assignment received from leader for group mycg for
generation 7 (kafka.coordinator.group.GroupCoordinator){noformat}
As a reference a regular consumer leave/shutdown logs looks like this. NOTE how
the consumer group moves into empty state.
{noformat}
[GroupCoordinator 0]: Member[group.instance.id None, member.id
consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7] in group mycg has left,
removing it from the group (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 8 (__consumer_offsets-22) (reason:
removing member consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7 on
LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Group mycg with generation 9 is now empty
(__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator){noformat}
was:
We've recently encountered a scenario where a consumer group got their
committed offsets deleted almost right after (around 3 minutes) the consumer
got into inactive state.
As per
[KIP-211|https://cwiki.apache.org/confluence/display/KAFKA/KIP-211%3A+Revise+Expiration+Semantics+of+Consumer+Group+Offsets],
committed offsets for an active (i.e running) consumer group should not be
deleted. However, if a consumer becomes inactive, {+}the deletion of committed
offsets will not occur immediately{+}. Instead, the committed offsets will only
be removed if the consumer remains inactive for at least the duration specified
by
[offset.retention.minutes|https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes].
In our case {{offset.retention.minutes}} is set to 7 days and the consumer was
only inactive for 5 minutes, so deletion should have not occurred.
By inspecting the KIP-211 further, we can find the following sentence:
{quote}If a group consumer unsubscribes from a topic but continues to consume
from other subscribed topics, the offset information of that unsubscribed
topic’s partitions should be deleted at the appropriate time.
{quote}
And later on:
{quote}If there are partitions the group has offset for but no longer consumes
from, and offsets.retention.minutes has passed since their last commit
timestamp, the corresponding offsets will be removed from the offset cache
{quote}
It is implied, though {*}+this is what I want to confirm in this ticket+{*},
that Kafka employs two approaches for offset expiration:
* The deletion timer is activated when a consumer group enters the Empty state
(i.e., not running). Once the timer exceeds the {{offset.retention.minutes}}
threshold, the committed offsets are deleted.
* If a consumer is in a "running" state (i.e., not in the Empty state) but is
no longer consuming from topics with committed offsets older than the
offset.retention.minutes duration, the committed offsets are deleted.
Note that the second approach only takes into account the timestamp of the last
committed offset.
Throughout this event, the affected consumer group didn’t transition into Empty
state. Based on the kafka logs, the consumer group was not detected as Empty,
indicating that Kafka considered the consumer to be running from its
perspective. It’s unclear why kafka didn’t detect this consumer group as Empty.
{noformat}
01:30:47 am - [GroupCoordinator 1]: Member consumer-mycg-1-uuid in group mycg
has failed, removing it from the group
01:30:47 am - [GroupCoordinator 1]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 432 (__consumer_offsets-16) (reason:
removing member consumer-mycg-1-uuid on heartbeat expiration)
1:30:50 am - [GroupCoordinator 1]: Member consumer-mycg-2-uuid in group mycg
has failed, removing it from the group
01:30:50 am - [GroupCoordinator 1]: Stabilized group mycg generation 433
(__consumer_offsets-16)
01:30:50 am - [GroupCoordinator 1]: Assignment received from leader for group
mycg for generation 433{noformat}
This suggests that kafka might have followed the second approach and that's why
kafka deleted the offsets 3 minutes later.
{noformat}
1:33:17 am -
[GroupMetadataManager brokerId=1] Removed 285 expired offsets in 8
milliseconds.{noformat}
As a reference a regular consumer join/startup logs looks like this. The group
is stabilised and the assignment from the leader received.
{noformat}
[GroupCoordinator 0]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 6 (__consumer_offsets-22) (reason:
Adding new member consumer-mycg-1-2b8ba689-fbaa-4829-82f5-dd2ed1d89d86 with
group instance id None) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Stabilized group mycg generation 7
(__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Assignment received from leader for group mycg for
generation 7 (kafka.coordinator.group.GroupCoordinator){noformat}
As a reference a regular consumer leave/shutdown logs looks like this. NOTE how
the consumer group moves into empty state.
{noformat}
[GroupCoordinator 0]: Member[group.instance.id None, member.id
consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7] in group mycg has left,
removing it from the group (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Preparing to rebalance group mycg in state
PreparingRebalance with old generation 8 (__consumer_offsets-22) (reason:
removing member consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7 on
LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
[GroupCoordinator 0]: Group mycg with generation 9 is now empty
(__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator){noformat}
As another piece of information, the consumer's underlying node experienced an
unclean termination, which could have played a role in Kafka's failure to
identify the consumer group as inactive.
In summary, when combining Kafka's expiration semantics of committed offsets
with Kafka's failure to detect a consumer in a dead state, it is possible for
committed offsets to be deleted.
> Consumer offsets can be deleted immediately if kafka does not detect a
> consumer as empty
> ----------------------------------------------------------------------------------------
>
> Key: KAFKA-15035
> URL: https://issues.apache.org/jira/browse/KAFKA-15035
> Project: Kafka
> Issue Type: Bug
> Affects Versions: 2.7.2
> Reporter: Sam Cantero
> Priority: Major
>
> We've recently encountered a scenario where a consumer group got their
> committed offsets deleted some minutes (around 3 minutes) after the consumer
> got into inactive state (the underlying node went away).
> As per
> [KIP-211|https://cwiki.apache.org/confluence/display/KAFKA/KIP-211%3A+Revise+Expiration+Semantics+of+Consumer+Group+Offsets],
> committed offsets for an active (i.e running) consumer group should not be
> deleted. However, if a consumer becomes inactive, {+}the deletion of
> committed offsets will not occur immediately{+}. Instead, the committed
> offsets will only be removed if the consumer remains inactive for at least
> the duration specified by
> [offset.retention.minutes|https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes].
> In our case {{offset.retention.minutes}} is set to 7 days and the consumer
> was only inactive for 5 minutes, so deletion should have not occurred.
> Later on KIP-496 was introduced to fix the following issue in KIP-211:
> {quote}When a consumer subscription changes, we are still left with the
> committed offsets of the previous subscription. These will never be cleaned
> up as long as the group remains active. We were aware of this problem in
> KIP-211, but the solution was not implemented because the coordinator is
> presently agnostic to join group metadata and we were unclear about the
> compatibility implications of changing that.
> {quote}
> However this introduced a regression as explained in
> https://issues.apache.org/jira/browse/KAFKA-13636.
> {quote}The group coordinator might delete invalid offsets during a group
> rebalance. During a rebalance, the coordinator is relying on the last commit
> timestamp ({_}offsetAndMetadata.commitTimestamp{_}) instead of the last state
> modification {_}timestamp (currentStateTimestamp{_}) to detect expired
> offsets.
> {quote}
> It is implied, that Kafka employs two approaches for offset expiration:
> * The deletion timer is activated when a consumer group enters the Empty
> state (i.e., not running). Once the timer exceeds the
> {{offset.retention.minutes}} threshold, the committed offsets are deleted.
> * If a consumer is in a "running" state (i.e., not in the Empty state) but
> is no longer consuming from topics with committed offsets older than the
> offset.retention.minutes duration, the committed offsets are deleted.
> However, the Kafka issue KAFKA-13636 specifically states that this situation
> could occur during a group rebalance. In my particular scenario, I have
> observed that the affected consumer group did not transition into the Empty
> state, and I have encountered a "Stabilized" log line. I'm uncertain whether
> this log line indicates that the rebalance had concluded at that point or
> not. The reason why Kafka did not detect this consumer group as Empty remains
> unclear.
> *Logs*
> {noformat}
> 01:30:47 am - [GroupCoordinator 1]: Member consumer-mycg-1-uuid in group mycg
> has failed, removing it from the group
> 01:30:47 am - [GroupCoordinator 1]: Preparing to rebalance group mycg in
> state PreparingRebalance with old generation 432 (__consumer_offsets-16)
> (reason: removing member consumer-mycg-1-uuid on heartbeat expiration)
> 1:30:50 am - [GroupCoordinator 1]: Member consumer-mycg-2-uuid in group mycg
> has failed, removing it from the group
> 01:30:50 am - [GroupCoordinator 1]: Stabilized group mycg generation 433
> (__consumer_offsets-16)
> 01:30:50 am - [GroupCoordinator 1]: Assignment received from leader for group
> mycg for generation 433{noformat}
> This suggests that kafka might have followed the second approach and that's
> why kafka deleted the offsets 3 minutes later.
> {noformat}
> 1:33:17 am -
> [GroupMetadataManager brokerId=1] Removed 285 expired offsets in 8
> milliseconds.{noformat}
> As a reference a regular consumer join/startup logs looks like this. The
> group is stabilised and the assignment from the leader received.
> {noformat}
> [GroupCoordinator 0]: Preparing to rebalance group mycg in state
> PreparingRebalance with old generation 6 (__consumer_offsets-22) (reason:
> Adding new member consumer-mycg-1-2b8ba689-fbaa-4829-82f5-dd2ed1d89d86 with
> group instance id None) (kafka.coordinator.group.GroupCoordinator)
>
> [GroupCoordinator 0]: Stabilized group mycg generation 7
> (__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator)
> [GroupCoordinator 0]: Assignment received from leader for group mycg for
> generation 7 (kafka.coordinator.group.GroupCoordinator){noformat}
> As a reference a regular consumer leave/shutdown logs looks like this. NOTE
> how the consumer group moves into empty state.
> {noformat}
> [GroupCoordinator 0]: Member[group.instance.id None, member.id
> consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7] in group mycg has left,
> removing it from the group (kafka.coordinator.group.GroupCoordinator)
> [GroupCoordinator 0]: Preparing to rebalance group mycg in state
> PreparingRebalance with old generation 8 (__consumer_offsets-22) (reason:
> removing member consumer-mycg-1-eb77a142-5b64-476e-bc3d-2731c9b811a7 on
> LeaveGroup) (kafka.coordinator.group.GroupCoordinator)
> [GroupCoordinator 0]: Group mycg with generation 9 is now empty
> (__consumer_offsets-22) (kafka.coordinator.group.GroupCoordinator){noformat}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
