[
https://issues.apache.org/jira/browse/KAFKA-19222?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-19222:
------------------------------
Component/s: clients
consumer
> Invalid FENCED_MEMBER_EPOCH error to ConsumerGroupHeartbeat
> -----------------------------------------------------------
>
> Key: KAFKA-19222
> URL: https://issues.apache.org/jira/browse/KAFKA-19222
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
> Affects Versions: 4.0.0, 4.0.1
> Reporter: Travis Bischel
> Priority: Major
>
> {{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)
