KAFKA-19183: Remove kafka.utils.Pool and refactor usages

[RaphaelFakhri]

Problem

The kafka.utils.Pool class is a thin Scala wrapper around java.util.concurrent.ConcurrentHashMap. While it provides a convenient getAndMaybePut method, the underlying functionality is largely replicated by ConcurrentHashMap's standard methods, particularly computeIfAbsent.

Analysis / Motivation

As discussed in KAFKA-19183, the Pool class adds an unnecessary layer of abstraction. The primary addition over raw ConcurrentHashMap is the getAndMaybePut method, which can be effectively replaced with ConcurrentHashMap.computeIfAbsent.

While Pool allows defining a default value factory at construction, analysis showed this feature is only used in a limited number of locations. These usages can be easily refactored to pass the value creation logic explicitly to computeIfAbsent at the call site.

Removing the Pool class:

• Simplifies the codebase
• Reduces maintenance overhead
• Promotes the use of standard, well-understood Java concurrent collections

Changes

Removal of Files

• kafka/utils/Pool.scala — Removed
• kafka/utils/PoolTest.scala — Removed (tested the now-removed Pool class)

Refactoring of Usages

• All instances of kafka.utils.Pool have been replaced with java.util.concurrent.ConcurrentHashMap.

• Calls to Pool.getAndMaybePut(key, createValue) have been replaced with:

  concurrentHashMap.computeIfAbsent(key, _ => createValue)

• Calls to Pool.getAndMaybePut(key) (which used the constructor factory) have been replaced with:

  concurrentHashMap.computeIfAbsent(key, k => valueFactory(k))

  where valueFactory is now locally defined or accessible in scope.

• Calls accessing collection views (.keys, .values, .iterator) have been updated to use Java API methods (.keySet(), .values(), .entrySet().iterator()) followed by .asScala from scala.jdk.CollectionConverters to maintain the Scala collection interface where necessary.

• Other minor method name updates:

  • .size → .size()
  • .putIfNotExists → .putIfAbsent

• The addLoadedTransactionsToCache method signature in TransactionStateManager has been updated to accept ConcurrentHashMap instead of Pool.

Testing

• PoolTest.scala was removed.
• Existing unit and integration tests for the components that previously used Pool (e.g., FetcherLagStats, TransactionStateManager, ReplicaManager) provide sufficient coverage for the refactored code paths.
• No new tests were added, as the core concurrent map functionality is provided by the JDK.

Compatibility

• This change is internal refactoring and does not affect any public APIs or external compatibility.

Checklist

• Code follows the coding style.
• Documentation is updated. (Not applicable for this internal refactoring)
• Unit tests are updated/added (Removed PoolTest, relying on existing component tests)
• Integration tests are updated/added. (Not applicable, existing tests cover functionality)
• KIP has been created or exists for this change. (KAFKA-19183 serves as the discussion/motivation)
• Release notes are updated. (Not applicable for this internal refactoring)

[FrankYang0529]

Hi @RaphaelFakhri, I will handle the issue KAFKA-19183. Could you check with Jira owner before creating PR next time? Thanks.

[frankvicky]

Hi @RaphaelFakhri
As @FrankYang0529 said, if you are willing to contribute, please refer to the contributing guide.
If you are willing to handle an issue that already has an assignee, please make sure you have asked the original assignee first.

[github-actions]

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.