Status

Current state: "Under Discussion"

Discussion thread: here

Vote thread: here

JIRA: KAFKA-19784

PR: https://github.com/apache/kafka/pull/20691

Motivation

Apache Kafka supports rack-aware partition assignment to improve fault tolerance and reduce cross-rack network traffic/cost. However, the current Admin API does not expose rack information for all type of consumer group members, despite this information being available at the protocol level.

Current State

The ConsumerGroupDescribeResponse (API Key 69) protocol includes a rackId field for each group member, as defined in the protocol specification:

{ "name": "RackId", "type": "string", "versions": "0+", 
  "nullableVersions": "0+", "default": "null",
  "about": "The member rack ID." }

However, when users call AdminClient.describeConsumerGroups(), the returned MemberDescription objects do not include this rack information. The rack ID is available in the wire protocol but is discarded during response processing in DescribeConsumerGroupsHandler.

The Similar case is for AdminClient.describeShareGroups():


Problem Statement

This limitation creates several issues:

1. Monitoring and Observability: Operators cannot determine the rack distribution of consumer group members through the Admin API, making it difficult to verify that rack-aware assignment is working correctly.

2. Inconsistency: Other group types expose rack information:

   • StreamsGroupMemberDescription includes rackId() method
   • The underlying protocol supports it for consumer groups
   • Only the public Admin API omits this information
3. Diagnostics: When troubleshooting rack-aware assignment issues or network problems, operators need to use lower-level tools or custom code to access rack information that should be readily available.
4. Third-party Tools: Monitoring and management tools built on the Admin API cannot display rack information, limiting their usefulness for rack-aware deployments.

Take one example:  currently we have to implemen our AZ/Rack analysis using a workaround — passing the rack information into the clientId field and parsing it afterward.

kafkaConsumerConfig.customConfig(ConsumerConfig.CLIENT_ID_CONFIG, generateClientIdWithRack(ip, rack));

Public Interfaces

add dedicated public interface for this case:

/**
 * Simple callback interface for broker ready notification.
 */
public interface BrokerReadyCallback {
    /**
     * This method will be called during broker startup for the implementation
     * which needs delayed initialization until the broker can process requests.
     */
    void onBrokerReady();

}

add the interface into RemoteLogMetadataManager's implements with default implement.

public interface RemoteLogMetadataManager extends BrokerReadyCallback, Configurable, Closeable 

Proposed Changes

 You can refer to https://github.com/apache/kafka/pull/20203/files

 We postpone the TopicBasedRemoteLogMetadataManager's initialization part (quering metedata from remote topic) after the server is ready for the request. 

Then the retry time is reasonable without worried about different kafka clusters. and the connection not available log won't be seen.

Compatibility, Deprecation, and Migration Plan 

• Considerations for compatibility：remote.log.metadata.initialization.retry.max.timeout.ms' implement changed
  Previously, the timer was started during TopicBasedRemoteLogMetadataManager initialization, before the broker was ready to handle requests.
  After this change, the timer starts after the broker is ready to handle requests.
  However, this change effectively increases the existing timeout duration. It does not break existing functionality and introduces no compatibility issues. The documentation for the time can be updated to clarify the starting point of the timer.
  
  

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

• Will cover the patch with deploy test and check if the startup can success without any connection retry error for remote storage. You can refer to test case

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

• Another discussed approach is not to call RLMM#configure() method while instantiating the RemoteLogManager#L422 and define a new method in RemoteLogManager#configureRLMM and this can be called from the BrokerServer. You can refer to the code.
  But the change breaks that contract:
  Accroding to KIP-877: "If a plugin implements this interface, the withPluginMetrics() method will be called when the plugin is instantiated (after configure() if the plugin also implements Configurable). "