Travis Bischel created KAFKA-19222:
--------------------------------------
Summary: Invalid FENCED_MEMBER_EPOCH error to
ConsumerGroupHeartbeat
Key: KAFKA-19222
URL: https://issues.apache.org/jira/browse/KAFKA-19222
Project: Kafka
Issue Type: Bug
Affects Versions: 4.0.0, 4.0.1
Reporter: Travis Bischel
{{throwIfConsumerGroupMemberEpochIsInvalid}} is meant to check the epoch in
heartbeat requests. If a ConsumerGroupHeartbeat response is lost (connection
dies before the client reads the response), the client may issue a
ConsumerGroupHeartbeat with the old epoch.
{{throwIfConsumerGroupMemberEpochIsInvalid}} means to allow this with this
conditional:
{code:java}
} else if (receivedMemberEpoch < member.memberEpoch()) {
// If the member comes with the previous epoch and has a subset of
the current assignment partitions,
// we accept it because the response with the bumped epoch may have
been lost.
if (receivedMemberEpoch != member.previousMemberEpoch() ||
!isSubset(ownedTopicPartitions, member.assignedPartitions())) {
throw new FencedMemberEpochException("The consumer group member
has a smaller member "
+ "epoch (" + receivedMemberEpoch + ") than the one known
by the group coordinator ("
+ member.memberEpoch() + "). The member must abandon all
its partitions and rejoin.");
}
}
{code}
However, {{isSubset}} immediately returns false if {{ownedPartitions}} is null.
Clients are not meant to send the {{ownedPartitions}} field
(ConsumerHeartbeatRequest.TopicPartitions) _unless_ the {{ownedPartitions}} has
changed.
As a concrete example, here are logs from a broker for a client. In this flow,
* ConsumerGroupHeartbeat is being issued within the client
* Concurrently, I am trying to leave the group -- this canceled the in-flight
ConsumerGroupHeartbeat from the client side
* I issue OffsetCommit (before the group is left)
* The OffsetCommit fails with STALE_MEMBER_EPOCH
* I issue a ConsumerGroupHeartbeat with _only_ the Group, MemberID, and
MemberEpoch fields set -- i.e. _nothing else is changing just give me the
latest epoch_
* Broker side, the broker finally processes my initial ConsumerGroupHeartbeat
that was initially sent (and already canceled client side) -- in the logs you
can see the epoch is bumped to 7 and the heartbeat is successful
* Broker side, the broker processes a _duplicate ConsumerGroupHeartbeat_ and
replies FENCED_MEMBER_EPOCH
* Client side, I bail on committing and finally leave the group. I send a
ConsumerGroupHeartbeat with memberEpoch -1.
* Broker side, the final heartbeat is processed successfully.
Broker side logs:
{code}
2025-04-30 17:22:26,024
[data-plane-kafka-network-thread-2-ListenerName(PLAINTEXT)-PLAINTEXT-2] DEBUG
kafka.request.logger - Completed
request:{"isForwarded":false,"requestHeader":{"requestApiKey":8,"requestApiVersion":9,"correlationId":0,"clientId":"kgo","requestApiKeyName":"OFFSET_COMMIT"},"request":{"groupId":"c24d1994653ddbaa5ff68d14c54c8733a7f02f732f4da4b5811f80ce0b247c2d","generationIdOrMemberEpoch":6,"memberId":"6j403VCp6yHIJrSlfWTtYg==","groupInstanceId":null,"topics":[{"name":"c5fd69bff34c4bde0f955fcf6de5d1cc2487067028e2383b6106804d3a9949b5","partitions":[{"partitionIndex":4,"committedOffset":45435,"committedLeaderEpoch":0,"committedMetadata":"6j403VCp6yHIJrSlfWTtYg=="},{"partitionIndex":3,"committedOffset":45371,"committedLeaderEpoch":0,"committedMetadata":"6j403VCp6yHIJrSlfWTtYg=="}]}]},"response":{"throttleTimeMs":0,"topics":[{"name":"c5fd69bff34c4bde0f955fcf6de5d1cc2487067028e2383b6106804d3a9949b5","partitions":[{"partitionIndex":4,"errorCode":113},{"partitionIndex":3,"errorCode":113}]}]},"connection":"127.0.0.1:9094-127.0.0.1:56328-2-197","totalTimeMs":175.381,"requestQueueTimeMs":0.027,"localTimeMs":0.037,"remoteTimeMs":175.075,"throttleTimeMs":0,"responseQueueTimeMs":0.118,"sendTimeMs":0.121,"securityProtocol":"PLAINTEXT","principal":"User:ANONYMOUS","listener":"PLAINTEXT","clientInformation":{"softwareName":"unknown","softwareVersion":"unknown"}}
2025-04-30 17:22:26,121
[data-plane-kafka-network-thread-2-ListenerName(PLAINTEXT)-PLAINTEXT-2] DEBUG
kafka.request.logger - Completed
request:{"isForwarded":false,"requestHeader":{"requestApiKey":68,"requestApiVersion":1,"correlationId":14,"clientId":"kgo","requestApiKeyName":"CONSUMER_GROUP_HEARTBEAT"},"request":{"groupId":"c24d1994653ddbaa5ff68d14c54c8733a7f02f732f4da4b5811f80ce0b247c2d","memberId":"6j403VCp6yHIJrSlfWTtYg==","memberEpoch":6,"instanceId":null,"rackId":null,"rebalanceTimeoutMs":-1,"subscribedTopicNames":null,"subscribedTopicRegex":null,"serverAssignor":null,"topicPartitions":null},"response":{"throttleTimeMs":0,"errorCode":0,"errorMessage":null,"memberId":"6j403VCp6yHIJrSlfWTtYg==","memberEpoch":7,"heartbeatIntervalMs":5000,"assignment":null},"connection":"127.0.0.1:9094-127.0.0.1:48238-2-153","totalTimeMs":273.886,"requestQueueTimeMs":0.021,"localTimeMs":0.025,"remoteTimeMs":273.594,"throttleTimeMs":0,"responseQueueTimeMs":0.062,"sendTimeMs":0.183,"securityProtocol":"PLAINTEXT","principal":"User:ANONYMOUS","listener":"PLAINTEXT","clientInformation":{"softwareName":"kgo","softwareVersion":"unknown"}}
2025-04-30 17:22:26,135
[data-plane-kafka-network-thread-2-ListenerName(PLAINTEXT)-PLAINTEXT-2] DEBUG
kafka.request.logger - Completed
request:{"isForwarded":false,"requestHeader":{"requestApiKey":68,"requestApiVersion":1,"correlationId":1,"clientId":"kgo","requestApiKeyName":"CONSUMER_GROUP_HEARTBEAT"},"request":{"groupId":"c24d1994653ddbaa5ff68d14c54c8733a7f02f732f4da4b5811f80ce0b247c2d","memberId":"6j403VCp6yHIJrSlfWTtYg==","memberEpoch":6,"instanceId":null,"rackId":null,"rebalanceTimeoutMs":-1,"subscribedTopicNames":null,"subscribedTopicRegex":null,"serverAssignor":null,"topicPartitions":null},"response":{"throttleTimeMs":0,"errorCode":110,"errorMessage":"The
consumer group member has a smaller member epoch (6) than the one known by the
group coordinator (7). The member must abandon all its partitions and
rejoin.","memberId":null,"memberEpoch":0,"heartbeatIntervalMs":0,"assignment":null},"connection":"127.0.0.1:9094-127.0.0.1:56328-2-197","totalTimeMs":3.304,"requestQueueTimeMs":0.095,"localTimeMs":0.091,"remoteTimeMs":2.67,"throttleTimeMs":0,"responseQueueTimeMs":0.306,"sendTimeMs":0.14,"securityProtocol":"PLAINTEXT","principal":"User:ANONYMOUS","listener":"PLAINTEXT","clientInformation":{"softwareName":"unknown","softwareVersion":"unknown"}}
2025-04-30 17:22:26,153
[data-plane-kafka-network-thread-2-ListenerName(PLAINTEXT)-PLAINTEXT-2] DEBUG
kafka.request.logger - Completed
request:{"isForwarded":false,"requestHeader":{"requestApiKey":68,"requestApiVersion":1,"correlationId":2,"clientId":"kgo","requestApiKeyName":"CONSUMER_GROUP_HEARTBEAT"},"request":{"groupId":"c24d1994653ddbaa5ff68d14c54c8733a7f02f732f4da4b5811f80ce0b247c2d","memberId":"6j403VCp6yHIJrSlfWTtYg==","memberEpoch":-1,"instanceId":null,"rackId":null,"rebalanceTimeoutMs":-1,"subscribedTopicNames":null,"subscribedTopicRegex":null,"serverAssignor":null,"topicPartitions":null},"response":{"throttleTimeMs":0,"errorCode":0,"errorMessage":null,"memberId":"6j403VCp6yHIJrSlfWTtYg==","memberEpoch":-1,"heartbeatIntervalMs":0,"assignment":null},"connection":"127.0.0.1:9094-127.0.0.1:56328-2-197","totalTimeMs":6.528,"requestQueueTimeMs":0.088,"localTimeMs":0.116,"remoteTimeMs":6.166,"throttleTimeMs":0,"responseQueueTimeMs":0.053,"sendTimeMs":0.102,"securityProtocol":"PLAINTEXT","principal":"User:ANONYMOUS","listener":"PLAINTEXT","clientInformation":{"softwareName":"unknown","softwareVersion":"unknown"}}
{code}
This can be very easily triggered by always issuing ConsumerGroupHeartbeat
twice, ignoring the first response, in any test that has rebalancing.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
