KAFKA-19159: Removed time based evictions for share sessions #19500

chirag-wadhwa5 · 2025-04-17T05:57:53Z

Currently the share session cache is desgined like the fetch session
cache. If the cache is full and a new share session is trying to get get
initialized, then the sessions which haven't been touched for more than
2minutes are evicted. This wouldn't be right for share sessions as the
members also hold locks on the acquired records, and session eviction
would mean theose locks will need to be dropped and the corresponding
records re-delivered. This PR removes the time based eviction logic for
share sessions.

Refer: KAFKA-19159

apoorvmittal10

Mostly looks good to me, some doubts.

apoorvmittal10 · 2025-04-17T15:16:28Z

server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java

-        } else if (now - lastUsedEntry.getKey().lastUsedMs() > evictionMs) {
-            ShareSession session = lastUsedEntry.getValue();
-            remove(session);
-            evictionsMeter.mark();


What about evictionsMeter metric i.e. where it will get updated in new design? Should it go in remove method now? Having said that, you were right that now there is actually no eviction per se i.e. no forceful eviction based on size hence we might not require the metrics. What do you think?

If not eviction then should we have ShareSessionRemovePerSec meter, will it be of any help?

Thanks for the review. There are no evictions but sessions are still dropped on client disconnections. I believe ShareSessionsDroppedPerSec would be a better name. We can target the question of where it will go in the PR that introduces the client connection drop logic. Will update that PR once this one gets merged. But yeah, that shouldn't go in the remove method because I don't think we need this metric to also include the sessions dropped due to Final Share Fetch request.

apoorvmittal10 · 2025-04-17T15:26:00Z

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

@@ -428,8 +428,8 @@ class BrokerServer(
      val fetchManager = new FetchManager(Time.SYSTEM, new FetchSessionCache(fetchSessionCacheShards))

      val shareFetchSessionCache : ShareSessionCache = new ShareSessionCache(
-        config.shareGroupConfig.shareGroupMaxGroups * config.groupCoordinatorConfig.shareGroupMaxSize,
-        KafkaBroker.MIN_INCREMENTAL_FETCH_SESSION_EVICTION_MS)
+        config.shareGroupConfig.shareGroupMaxGroups * config.groupCoordinatorConfig.shareGroupMaxSize


We will replace this with new configuration in subsequent PRs, correct?

Thanks for the review. A PR is already in review to take care of this.
#19505

apoorvmittal10 · 2025-04-17T15:29:31Z

server/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java

-        TestUtils.waitForCondition(() -> yammerMetricValue(ShareSessionCache.SHARE_SESSIONS_COUNT).intValue() == 3,
-            "Share session count should be 3.");
-        TestUtils.waitForCondition(() -> yammerMetricValue(ShareSessionCache.SHARE_PARTITIONS_COUNT).intValue() == 60,
-            "Share partition count should be 60.");
-        assertEquals(0, cache.evictionsMeter().count());


Why session and partitions count metric is removed in test?

Thanks for the review. These haven't been removed. I have put these assertions in a separate method assertMetricsValues because these checks are repeated in the tests.

apoorvmittal10 · 2025-04-17T15:30:11Z

server/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java

-        assertEquals(0, cache.evictionsMeter().count());
-
-        // Touch the sessions to update the last used time, so that the key-2 can be evicted.
-        cache.touch(cache.get(key1), 200);


Shaouldn't we use cache.updateNumPartitions here?

Thanks for the review. This test actually touched the cache so that inactive sessions could be evicted and new ones could be added to the cache. But the eviction logic is removed now so replacing touch with updateNumPartitions wouldn't pass the test. Also since the resizing of cache session is being tested in the next test in this file, I removed the other logic to let this one only test the addition of sessions to the cache

apoorvmittal10

LGTM, can you please create a jira to track the change for evictions metric. EIther we replace or drop but should be tracked.

chia7712 · 2025-04-17T18:36:12Z

server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java

     */
-    public synchronized void touch(ShareSession session, long now) {
+    public synchronized void updateNumPartitions(ShareSession session) {
        synchronized (session) {


Pardon me, why to have this sync? it seems the methods cachedSize and cachedSize are under synchronized already.

Thanks for the review. Didn't notice that before, I have made the required changes in the latest commit.

There is a read and update to the session, to make the operation atomic over ShareSession hence the sync block was applied here. The individual methods sync can guarantee an updated value value read but overall the operation is not atomic if we miss the sync block here. Am I missing something?

There is a read and update to the session

You are right. Maybe we can add a specific method to handle the case to ensure the atomicity? for example:

ShareSession

public synchronized int updateCachedSize() { var previousSize = cachedSize; cachedSize = partitionMap.size(); return previousSize != -1 ? cachedSize - previousSize : cachedSize; }

ShareSessionCache

public synchronized void updateNumPartitions(ShareSession session) { numPartitions += session.updateCachedSize(); }

Yes, the above also works or a sync block as earlier. I incline towards the former, which you suggested.

I have made the updates to the code
cc - @apoorvmittal10

chia7712 · 2025-04-17T18:40:53Z

server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java

@@ -122,15 +113,9 @@ public synchronized ShareSession remove(ShareSession session) {
     * Update a session's position in the lastUsed tree.


please update the docs

…reSessionCache

…artitions method

apoorvmittal10

Thanks for the changes, minor comment.

apoorvmittal10 · 2025-04-21T15:01:13Z

server/src/main/java/org/apache/kafka/server/share/session/ShareSession.java

@@ -138,6 +118,12 @@ public synchronized Map<ModifiedTopicIdPartitionType, List<TopicIdPartition>> up
        return result;
    }

+    public synchronized int updateCachedSize() {


This method needs javadoc comments saying specially about the return data and what the return value represents.

apoorvmittal10

LGTM! I will wait for sometime if others have any further comments.

KAFKA-19159: Removed time based evictions for share sessions

396d277

github-actions bot added triage PRs from the community core Kafka Broker KIP-932 Queues for Kafka labels Apr 17, 2025

m1a2st added the ci-approved label Apr 17, 2025

apoorvmittal10 reviewed Apr 17, 2025

View reviewed changes

chirag-wadhwa5 requested a review from apoorvmittal10 April 17, 2025 16:48

apoorvmittal10 approved these changes Apr 17, 2025

View reviewed changes

chia7712 reviewed Apr 17, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Apr 18, 2025

KAFKA-19159: removed synchronized block in updateNumPartitions in Sha…

321582e

…reSessionCache

chirag-wadhwa5 requested review from chia7712 and apoorvmittal10 April 21, 2025 05:28

chirag-wadhwa5 added 3 commits April 21, 2025 15:25

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19159

6f388ad

KAFKA-19159: added synchronized block in shareSessionCache updateNumP…

5fe028a

…artitions method

KAFKA-19159: minor changes for better concurrency control

fc1d648

apoorvmittal10 requested changes Apr 21, 2025

View reviewed changes

KAFKA-19159: Added java docs for updateCachedSize in ShareSession.java

f9ad511

chirag-wadhwa5 requested a review from apoorvmittal10 April 21, 2025 15:18

apoorvmittal10 approved these changes Apr 21, 2025

View reviewed changes

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19159

a1aeceb

apoorvmittal10 merged commit 22c5794 into apache:trunk Apr 22, 2025
20 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-19159: Removed time based evictions for share sessions #19500

KAFKA-19159: Removed time based evictions for share sessions #19500

chirag-wadhwa5 commented Apr 17, 2025 •

edited by github-actions bot

Loading

apoorvmittal10 left a comment

apoorvmittal10 Apr 17, 2025

apoorvmittal10 Apr 17, 2025

chirag-wadhwa5 Apr 17, 2025

apoorvmittal10 Apr 17, 2025

chirag-wadhwa5 Apr 17, 2025

apoorvmittal10 Apr 17, 2025

chirag-wadhwa5 Apr 17, 2025

apoorvmittal10 Apr 17, 2025

chirag-wadhwa5 Apr 17, 2025

apoorvmittal10 left a comment

chia7712 Apr 17, 2025

chirag-wadhwa5 Apr 21, 2025

apoorvmittal10 Apr 21, 2025

chia7712 Apr 21, 2025

apoorvmittal10 Apr 21, 2025 •

edited

Loading

chirag-wadhwa5 Apr 21, 2025

chia7712 Apr 17, 2025

apoorvmittal10 left a comment

apoorvmittal10 Apr 21, 2025

apoorvmittal10 left a comment

		@@ -122,15 +113,9 @@ public synchronized ShareSession remove(ShareSession session) {
		* Update a session's position in the lastUsed tree.

KAFKA-19159: Removed time based evictions for share sessions #19500

KAFKA-19159: Removed time based evictions for share sessions #19500

Conversation

chirag-wadhwa5 commented Apr 17, 2025 • edited by github-actions bot Loading

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 Apr 21, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 left a comment

Choose a reason for hiding this comment

chirag-wadhwa5 commented Apr 17, 2025 •

edited by github-actions bot

Loading

apoorvmittal10 Apr 21, 2025 •

edited

Loading