chirag-wadhwa5 commented on code in PR #19329:
URL: https://github.com/apache/kafka/pull/19329#discussion_r2042084615
##########
server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java:
##########
@@ -140,50 +124,46 @@ public synchronized void touch(ShareSession session, long
now) {
}
}
- /**
- * Try to evict an entry from the session cache.
- * <p>
- * A proposed new element A may evict an existing element B if:
- * B is considered "stale" because it has been inactive for a long time.
- *
- * @param now The current time in milliseconds.
- * @return True if an entry was evicted; false otherwise.
- */
- public synchronized boolean tryEvict(long now) {
- // Try to evict an entry which is stale.
- Map.Entry<LastUsedKey, ShareSession> lastUsedEntry =
lastUsed.firstEntry();
- if (lastUsedEntry == null) {
- return false;
- } else if (now - lastUsedEntry.getKey().lastUsedMs() > evictionMs) {
- ShareSession session = lastUsedEntry.getValue();
- remove(session);
- evictionsMeter.mark();
- return true;
- }
- return false;
- }
-
/**
* Maybe create a new session and add it to the cache.
* @param groupId - The group id in the share fetch request.
* @param memberId - The member id in the share fetch request.
* @param now - The current time in milliseconds.
* @param partitionMap - The topic partitions to be added to the session.
+ * @param clientConnectionId - The client connection id.
* @return - The session key if the session was created, or null if the
session was not created.
*/
- public synchronized ShareSessionKey maybeCreateSession(String groupId,
Uuid memberId, long now, ImplicitLinkedHashCollection<CachedSharePartition>
partitionMap) {
- if (sessions.size() < maxEntries || tryEvict(now)) {
+ public synchronized ShareSessionKey maybeCreateSession(
+ String groupId,
+ Uuid memberId,
+ long now,
+ ImplicitLinkedHashCollection<CachedSharePartition> partitionMap,
+ String clientConnectionId
+ ) {
+ if (sessions.size() < maxEntries) {
ShareSession session = new ShareSession(new
ShareSessionKey(groupId, memberId), partitionMap,
now, now,
ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH));
sessions.put(session.key(), session);
touch(session, now);
Review Comment:
Thanks for the review. I have changed the touch method to
updateNumPartitions as we still need to update the number of share partitions
in the session
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
