MINOR: tidy up SocketServerMemoryPoolTest by edoardocomar · Pull Request #21873 · apache/kafka

edoardocomar · 2026-03-26T02:42:45Z

moved to package org.apache.kafka.common
use IntegrationTestUtils and RequestUtils to create messages
made SocketServer MemoryPool accessible for testing

following up comments in #21740

note that the MemoryPoolAvailable metrics is registered in the singleton
default Yammer registry and it is not scoped per listener because there
is one pool per SocketServer, not one per listener. So it is not usable
in this ClusterTest.

Reviewers: Mickael Maison mimaison@users.noreply.github.com, Mickael
Maison mickael.maison@gmail.com

github-actions · 2026-04-02T04:04:41Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

mimaison · 2026-04-08T12:20:02Z

  memoryPoolSensor.add(new Meter(TimeUnit.MILLISECONDS, memoryPoolDepletedPercentMetricName, memoryPoolDepletedTimeMetricName))
-  private val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE
+  // accessible for testing
+  val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE


You said that the SocketServer metrics collide when using ClusterTest.
It seems like it would also be the case if you start a single server in combined mode, right?

yes, there are two SocketServers instances, one for BrokerServer one fo ControllerServer, and each of them defines
metricsGroup.newGauge("MemoryPoolAvailable", () => memoryPool.availableMemory)
so the singleton Yammerregsitry only contains one gauge for that name,
so it cannot be used to read the broker's pool availableMemory - so i went for accessing the value directly in the test, not via the metric

My point is that it's a bug. If you run a server in combined mode, you have 2 SocketServer instances each with its own MemoryPool but there's a single metric. Since newGauge() is a "get or create", I'm guessing the metric reports the value for the component that create its metric first. If so we should try to fix it (not in this PR).

agreed @mimaison - I opened https://issues.apache.org/jira/browse/KAFKA-20453

as for the ClusterTest, even in normal KRaft you get multiple SocketServers instances in a single JVM and a single metric.

Would you consider approving this minor patch and leave KAFKA-20453 to another PR?
Although i am not sure for how long the scala SocketServer and KafkaYammerMetrics singleton will be around!

mimaison

Thanks for the PR. It's a nice cleanup, I left a few small comments

mimaison · 2026-05-07T16:27:17Z

-        // Header: api_key(2) + api_version(2) + correlation_id(4) + client_id_len(2) + client_id
-        int headerSize = 2 + 2 + 4 + 2 + clientIdBytes.length;
-        int payloadSize = headerSize + body.length;
+    private int getBrokerBoundPort(ClusterInstance clusterInstance) {


If there's a good reason why we can't use brokerBoundPorts(), we have the boundPort() method on BrokerServer to retrieve the port. That method uses socketServer itself so we don't have to.

mimaison · 2026-05-07T16:38:07Z

        long initialMemoryPoolAvailable = getMemoryPoolAvailable(clusterInstance);

-        try (Socket socket = IntegrationTestUtils.connect(clusterInstance.brokerBoundPorts().get(0))) {
+        try (Socket socket = IntegrationTestUtils.connect(getBrokerBoundPort(clusterInstance))) {


Why can't we use clusterInstance.brokerBoundPorts().get(0) anymore?

mimaison · 2026-05-07T16:46:33Z

-        int headerSize = 2 + 2 + 4 + 2 + clientIdBytes.length;
-        int payloadSize = headerSize + body.length;
+    private int getBrokerBoundPort(ClusterInstance clusterInstance) {
+        return getSocketServer(clusterInstance).boundPort(ListenerName.normalised(TestKitDefaults.DEFAULT_BROKER_LISTENER_NAME));


Again if we can't use brokerBoundPorts(), we should use clusterInstance.clientListener() to retrieve the listener name.

edoardocomar · 2026-05-22T16:48:10Z

thanks for your comments, I've addressed your suggestions

mimaison

Thanks for the updates, I think it's a nice cleanup. I left a small suggestion

mimaison · 2026-05-26T11:03:46Z

@@ -98,7 +94,8 @@ class SocketServer(
  private val memoryPoolDepletedPercentMetricName = metrics.metricName("MemoryPoolAvgDepletedPercent", JSocketServer.METRICS_GROUP)
  private val memoryPoolDepletedTimeMetricName = metrics.metricName("MemoryPoolDepletedTimeTotal", JSocketServer.METRICS_GROUP)
  memoryPoolSensor.add(new Meter(TimeUnit.MILLISECONDS, memoryPoolDepletedPercentMetricName, memoryPoolDepletedTimeMetricName))
-  private val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE
+  // accessible for testing
+  val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE


Nit: I think we could make this private[network] instead of fully exposing it.

Done, thanks

TIL

private[network] in Scala 2 compiles to a public getter method in JVM bytecode. The private[network] restriction is enforced only by the Scala compiler — it will reject Scala code outside kafka.network that tries to call memoryPool(). But the underlying bytecode method has public visibility, so Java code can call
it freely, completely bypassing the Scala-level restriction.

This is a fundamental limitation of how Scala 2 maps its package-scoped visibility to the JVM: the JVM has no concept of "private to a specific named package," so the compiler can only enforce the restriction syntactically during Scala-to-Scala compilation. Any Java caller (or even reflection) sees a fully public method.

Also TIL, I knew it worked but never considered why. Thanks

edoardocomar · 2026-05-27T14:35:17Z

@mimaison after rebasing to trunk before pushing I encountered a thread leak failure
which for me exists in trunk already, irrespective of this PR, so i opened

https://issues.apache.org/jira/browse/KAFKA-20619

and PR
#22386

edoardocomar · 2026-05-27T23:50:13Z

@mimaison the CI builds fail with something unrelated to this PR

Error: The action gradle/actions/setup-gradle@4d9f0ba0025fe599b4ebab900eb7f3a1d93ef4c2 is not allowed in apache/kafka because all actions must be from a repository owned by your enterprise, created by GitHub, or match one of the patterns: 1Password/load-secrets-action/configure@8d0d610af187e78a2772c2d18d627f4c52d3fbfb, 1Password/load-secrets-action/configure@92467eb28f72e8255933372f1e0707c567ce2259, 1Password/load-secrets-action/configure@dafbe7cb03502b260e2b2893c753c352eee545bf, 1Password/load-secrets-action@8d0d610af187e78a2772c2d18d627f4c52d3fbfb, 1Password/load-secrets-action@92467eb28f72e8255933372f1e0707c567ce2259, 1Password/load-secrets-action@dafbe7cb03502b260e2b2893c753c352eee545bf, AdoptOpenJDK/install-jdk@*, BobAnkh/auto-generate-changelog@*, DavidAnson/markdownlint-cli2-action@07035fd053f7be764496c0f8d8f9f41f98305101, DavidAnson/markdownlint-cli2-action@ce4853d43830c74c1753b39f3cf40f71c2031eb9, DavidAnson/markdownlint-cli2-action@ded1f9488f68a970bc66ea5619e13e9b52e601cd, Enrico...

probably i just need to rebase

* moved to package org.apache.kafka.common * use IntegrationTestUtils and RequestUtils to create messages * made SocketServer MemoryPool accessible for testing

github-actions Bot added the triage PRs from the community label Mar 26, 2026

edoardocomar requested a review from mimaison March 26, 2026 02:42

github-actions Bot added core Kafka Broker clients small Small PRs labels Mar 26, 2026

edoardocomar force-pushed the ec/KAFKA-20302-test-tidy branch 2 times, most recently from 71d21c7 to 7f38206 Compare March 26, 2026 02:48

edoardocomar requested a review from showuon March 26, 2026 11:49

github-actions Bot added the needs-attention label Apr 2, 2026

mimaison reviewed Apr 8, 2026

View reviewed changes

github-actions Bot removed needs-attention triage PRs from the community labels Apr 9, 2026

ChoMinGi mentioned this pull request Apr 16, 2026

KAFKA-20453: Add nodeId and listenerType tags to SocketServer metrics #22074

Closed

mimaison reviewed May 7, 2026

View reviewed changes

edoardocomar force-pushed the ec/KAFKA-20302-test-tidy branch from 9f23d39 to e6e51b4 Compare May 22, 2026 17:04

mimaison approved these changes May 26, 2026

View reviewed changes

edoardocomar and others added 5 commits May 28, 2026 00:54

MINOR: tidy up SocketServerMemoryPoolTest

7d1f6bd

* moved to package org.apache.kafka.common * use IntegrationTestUtils and RequestUtils to create messages * made SocketServer MemoryPool accessible for testing

helper methods

df01205

spotlessApply

b57c3fe

implemented review suggestions

dc728a2

implemented review suggestion

f2b75ee

edoardocomar force-pushed the ec/KAFKA-20302-test-tidy branch from 4dcc5e2 to f2b75ee Compare May 28, 2026 00:01

edoardocomar merged commit 5c93ec9 into apache:trunk May 28, 2026
20 checks passed

edoardocomar deleted the ec/KAFKA-20302-test-tidy branch May 28, 2026 08:44

Conversation

edoardocomar commented Mar 26, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Apr 2, 2026

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!

edoardocomar Apr 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mimaison 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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar commented May 22, 2026

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar commented May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

edoardocomar commented May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

edoardocomar commented Mar 26, 2026 •

edited by github-actions Bot

Loading

edoardocomar Apr 15, 2026 •

edited

Loading

edoardocomar May 27, 2026 •

edited

Loading

edoardocomar commented May 27, 2026 •

edited

Loading

edoardocomar commented May 27, 2026 •

edited

Loading