KAFKA-19800-2: Created an implementation class NetworkPartitionMetadataClient for PartitionMetadataClient #20852

chirag-wadhwa5 · 2025-11-10T10:42:40Z

This PR is part of
KIP-1226.

This PR introduces an implementation class
NetworkPartitionMetadataClient for PartitionMetadataClient, that uses a
NetworkClient to send ListOffsetsRequest to the destination node. The
destination node should be the leader broker for the partitions in the
request and is retrieved from MetadataCache.

This new imple class will later be used in GroupCoordinatorService to
find the partition end offsets while computing share partition lag for
DescribeShareGroupOffsets request.

…or fetching partition end offset

apoorvmittal10

Thanks for the PR, some basic questions in the comments. Also can you please add the change for the usage of PartitionMetadataClient from where it will be initialized.

apoorvmittal10 · 2025-11-10T19:03:06Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        Set<TopicPartition> topicPartitions
+    ) {
+        if (topicPartitions == null || topicPartitions.isEmpty()) {
+            return new HashMap<>();


Suggested change

return new HashMap<>();

return Map.of();

apoorvmittal10 · 2025-11-10T19:20:26Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        this.sendThread = new SendThread(
+            "NetworkPartitionMetadataClientSendThread",
+            networkClient,
+            Math.toIntExact(CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS),  //30 seconds
+            this.time
+        );
+        this.sendThread.start();


Are we going to crerate a new NetworkClient with new connection to every broker or re-use some existing network client?

Can we have the usgae of this class in the same PR?

Again if it needs to be lazily loaded then we might have to delay the start.

AndrewJSchofield

Thanks for the PR. It looks pretty complete with just a few comments.

AndrewJSchofield · 2025-11-11T09:52:36Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+            return requests;
+        }
+
+        private void handleErrorResponse(PendingRequest pendingRequest, ClientResponse clientResponse) {


nit: I'd put this method beneath handleResponse just for ease of reading and maintenance.

AndrewJSchofield · 2025-11-11T09:54:30Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        }
+    }
+
+    private ListOffsetsPartitionResponse createErrorPartitionResponse(TopicPartition tp, short errorCode) {


nit: Let's put this utility method right at the end of the class.

AndrewJSchofield · 2025-11-11T10:10:02Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+        assertTrue(futures.containsKey(tp));
+
+        ListOffsetsPartitionResponse response = futures.get(tp).get();
+        assertNotNull(response);


Let's have an assertion that the future is completed in all of these tests please.

AndrewJSchofield · 2025-11-11T10:14:02Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+        ListOffsetsPartitionResponse response = futures.get(tp).get();
+        assertNotNull(response);
+        assertEquals(PARTITION, response.partitionIndex());
+        assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION.code(), response.errorCode());


Let's check that the offset, timestamp and epoch are -1 in the responses for the error cases.

AndrewJSchofield · 2025-11-11T10:15:14Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+                            new ListOffsetsPartitionResponse()
+                                .setPartitionIndex(PARTITION)
+                                .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
+                                .setOffset(ListOffsetsResponse.UNKNOWN_OFFSET)


Because -1 are the default values here, I think you could not set the values in this constructor. In this way, you would be testing that error paths which uses this response and do not set values do actually pick up the defaults with no extra effort.

apoorvmittal10 · 2025-11-11T15:30:06Z

...inator/src/main/java/org/apache/kafka/coordinator/group/metrics/PartitionMetadataClient.java

+/**
+ * Client interface for retrieving latest offsets for topic partitions.
+ */
+public interface PartitionMetadataClient extends AutoCloseable {


You shouldn't need this interface now as the other PR is merged.

apoorvmittal10 · 2025-11-11T15:30:31Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        this.sendThread = new SendThread(
+            "NetworkPartitionMetadataClientSendThread",
+            networkClient,
+            Math.toIntExact(CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS),  //30 seconds
+            this.time
+        );
+        this.sendThread.start();


Can we have the usgae of this class in the same PR?

…tLatestOffsets

apoorvmittal10

Thanks for the PR, some comments.

apoorvmittal10 · 2025-11-12T18:03:27Z

core/src/main/scala/kafka/server/BrokerServer.scala

+      NetworkUtils.buildNetworkClient(
+        "NetworkPartitionMetadataClient",
+        config,
+        metrics,
+        Time.SYSTEM,
+        new LogContext(s"[NetworkPartitionMetadataClient broker=${config.brokerId}]")
+      ),


We need to decide if we want the PartitionMetadataClient to be lazily loaded, if yes then we should pass a provider which can fetch the NetworkClient.

I think this is a good model since it's not going to be used in every cluster.

apoorvmittal10 · 2025-11-12T18:03:39Z

core/src/main/scala/kafka/server/BrokerServer.scala

-
-      override def close(): Unit = {}
-    }
+  private def createNetworkPartitionMetadataClient(metadataCache: MetadataCache): PartitionMetadataClient = {


nit:

Suggested change

private def createNetworkPartitionMetadataClient(metadataCache: MetadataCache): PartitionMetadataClient = {

private def createPartitionMetadataClient(metadataCache: MetadataCache): PartitionMetadataClient = {

apoorvmittal10 · 2025-11-12T18:06:29Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        this.sendThread = new SendThread(
+            "NetworkPartitionMetadataClientSendThread",
+            networkClient,
+            Math.toIntExact(CommonClientConfigs.DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS),  //30 seconds
+            this.time
+        );
+        this.sendThread.start();


Again if it needs to be lazily loaded then we might have to delay the start.

apoorvmittal10 · 2025-11-12T18:09:49Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+
+        // Map to store futures for each TopicPartition
+        Map<TopicPartition, CompletableFuture<OffsetResponse>> futures = new HashMap<>();
+


nit: remove line break

apoorvmittal10 · 2025-11-12T18:11:14Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+
+            if (leaderNodeOpt.isEmpty() || leaderNodeOpt.get().isEmpty()) {
+                // No leader available - complete with error
+


remove line break

apoorvmittal10 · 2025-11-12T19:52:25Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+                return;
+            }
+
+            log.debug("ListOffsets response received - {}", clientResponse);


Why not this is the first line in the method?

Writing the log only during success scenarios might help better while debugging future issues, rather than logging it everytime. I have changed the log statement to ListOffsets response received successfully, for better understanding

apoorvmittal10 · 2025-11-12T19:52:39Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+            }
+
+            log.debug("ListOffsets response received - {}", clientResponse);
+


nit: remove line break

apoorvmittal10 · 2025-11-12T19:53:07Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+
+            log.debug("ListOffsets response received - {}", clientResponse);
+
+            // Parse the response


Is this comment providing any meaningful insight?

apoorvmittal10 · 2025-11-12T20:02:02Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+     */
+    private record PendingRequest(
+        Node node,
+        List<TopicPartition> partitions,


If these are only required to log in case of error then you can write pendingRequest.futures.keySet() and avoid passing List<TopicPartition> partitions altogether.

apoorvmittal10 · 2025-11-12T20:07:58Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+                for (ListOffsetsPartitionResponse partitionResponse : topicResponse.partitions()) {
+                    TopicPartition tp = new TopicPartition(topicName, partitionResponse.partitionIndex());
+                    // Remove the corresponding future from the map and complete it.
+                    CompletableFuture<OffsetResponse> future = pendingRequest.futures.remove(tp);


I do get why do you want to remove the topicPartition from the futures so can complete the pending ones later, but it's not a good idea to modify the futures list, as it can be passed as unmodifiable by some code in future. Hence, either keep a set of topic partitions which are completed and then while iterating pendingRequest.futures check if it's already contained in the set or a better one is to avoid set and just check if the future is still pending i.e.

pendingRequest.futures.forEach((tp, future) -> { // If future is not completed yet hence topic-partition was not included in the response, complete with error if (!future.isDone()) { future.complete(new OffsetResponse(-1, Errors.UNKNOWN_TOPIC_OR_PARTITION)); } });

apoorvmittal10

Thanks for the changes, looks good mostly. Some comments.

apoorvmittal10 · 2025-11-13T17:00:01Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+            return this;
+        }
+
+        public static NetworkPartitionMetadataClientBuilder bulider() {


Suggested change

public static NetworkPartitionMetadataClientBuilder bulider() {

static NetworkPartitionMetadataClientBuilder builder() {

apoorvmittal10 · 2025-11-13T17:00:36Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+            return new NetworkPartitionMetadataClientBuilder();
+        }
+
+        public NetworkPartitionMetadataClient build() {


Suggested change

public NetworkPartitionMetadataClient build() {

NetworkPartitionMetadataClient build() {

apoorvmittal10 · 2025-11-13T17:01:36Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+        NetworkPartitionMetadataClientBuilder withTime(Time time) {
+            this.time = time;
+            return this;


Are these methods used ever? If not then remove them.

apoorvmittal10 · 2025-11-13T17:08:52Z

core/src/test/java/kafka/server/NetworkPartitionMetadataClientTest.java

+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class NetworkPartitionMetadataClientTest {


Missing tests

Failure handling of condition: clientResponse.wasTimedOut() and clientResponse == null

Close method error condition - catch (InterruptedException e)

apoorvmittal10 · 2025-11-13T17:15:04Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+
+    private final MetadataCache metadataCache;
+    private final Supplier<KafkaClient> networkClientSupplier;
+    private volatile SendThread sendThread;


nit: move after the final variable declaration

apoorvmittal10 · 2025-11-13T17:15:12Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        this.networkClientSupplier = networkClientSupplier;
+        this.time = time;
+        this.listenerName = listenerName;
+        this.sendThread = null;


nit: not required

apoorvmittal10 · 2025-11-13T17:17:03Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+        if (sendThread == null) {
+            synchronized (initializationLock) {
+                if (sendThread == null) {


Better to use AtomicBoolean with compareAndSet.

apoorvmittal10 · 2025-11-13T17:17:33Z

core/src/main/java/kafka/server/NetworkPartitionMetadataClient.java

+                    );
+                    thread.start();
+                    sendThread = thread;
+                    log.debug("NetworkPartitionMetadataClient sendThread initialized and started");


Suggested change

log.debug("NetworkPartitionMetadataClient sendThread initialized and started");

log.info("NetworkPartitionMetadataClient sendThread initialized and started");

apoorvmittal10 · 2025-11-13T17:19:33Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

                            } catch (CompletionException e) {
                                // If fetching latest offset for a partition failed, return the error in the response for that partition.
                                partitionResponses.add(new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponsePartition()
                                    .setPartitionIndex(partitionData.partition())
                                    .setErrorCode(Errors.forException(e.getCause()).code())
                                    .setErrorMessage(e.getCause().getMessage()));
                            }


Do we need catch block now?

chirag-wadhwa5 added 3 commits November 7, 2025 16:10

Introduced PartitionMetadataClient and AdminPartitionMetadataClient f…

d2fb24e

…or fetching partition end offset

KAFKA-19800-2: Minor changes for better readability

37fb663

KAFKA-19800-2: Added tests for NetworkPartitionMetadataClientTest

19c8919

github-actions bot added triage PRs from the community core Kafka Broker group-coordinator labels Nov 10, 2025

AndrewJSchofield added KIP-932 Queues for Kafka ci-approved and removed triage PRs from the community labels Nov 10, 2025

KAFKA-19800-2: minor checkstyle changes

cab5eb1

chirag-wadhwa5 mentioned this pull request Nov 11, 2025

KAFKA-19800: Compute share partition lag in GroupCoordinatorService #20839

Merged

apoorvmittal10 reviewed Nov 11, 2025

View reviewed changes

AndrewJSchofield requested changes Nov 11, 2025

View reviewed changes

chirag-wadhwa5 added 2 commits November 11, 2025 20:53

KAFKA-19800-2: minor changes for better readability

b5091cf

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19800-2

5d241da

apoorvmittal10 requested changes Nov 11, 2025

View reviewed changes

KAFKA-19800-2: Changed the return type of PartitionMetadataClient.lis…

d009743

…tLatestOffsets

chirag-wadhwa5 requested review from AndrewJSchofield and apoorvmittal10 November 12, 2025 10:18

apoorvmittal10 requested changes Nov 12, 2025

View reviewed changes

chirag-wadhwa5 added 2 commits November 13, 2025 16:01

KAFKA-19800-2: Minor changes for better readability

b16ad34

KAFKA-19800-2: Loaded NetworkClient lazily

b541523

chirag-wadhwa5 requested a review from apoorvmittal10 November 13, 2025 17:08

apoorvmittal10 requested changes Nov 13, 2025

View reviewed changes

	private def createNetworkPartitionMetadataClient(metadataCache: MetadataCache): PartitionMetadataClient = {
	private def createPartitionMetadataClient(metadataCache: MetadataCache): PartitionMetadataClient = {


		// Map to store futures for each TopicPartition
		Map<TopicPartition, CompletableFuture<OffsetResponse>> futures = new HashMap<>();


		if (leaderNodeOpt.isEmpty() \|\| leaderNodeOpt.get().isEmpty()) {
		// No leader available - complete with error

		}

		log.debug("ListOffsets response received - {}", clientResponse);


		log.debug("ListOffsets response received - {}", clientResponse);

		// Parse the response

	public static NetworkPartitionMetadataClientBuilder bulider() {
	static NetworkPartitionMetadataClientBuilder builder() {

	public NetworkPartitionMetadataClient build() {
	NetworkPartitionMetadataClient build() {

	log.debug("NetworkPartitionMetadataClient sendThread initialized and started");
	log.info("NetworkPartitionMetadataClient sendThread initialized and started");

KAFKA-19800-2: Created an implementation class NetworkPartitionMetadataClient for PartitionMetadataClient #20852

Are you sure you want to change the base?

KAFKA-19800-2: Created an implementation class NetworkPartitionMetadataClient for PartitionMetadataClient #20852

Conversation

chirag-wadhwa5 commented Nov 10, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

chirag-wadhwa5 commented Nov 10, 2025 •

edited by github-actions bot

Loading