(ignite-3) branch main updated: IGNITE-25385 Make TransactionExpirationRegistry thread-safe (#5819)

[rp...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rpuch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git


The following commit(s) were added to refs/heads/main by this push:
     new f1730325375 IGNITE-25385 Make TransactionExpirationRegistry thread-safe (#5819)
f1730325375 is described below

commit f1730325375d3c0886f8b69511e568a4bfe3eb23
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Thu May 15 21:48:30 2025 +0400

    IGNITE-25385 Make TransactionExpirationRegistry thread-safe (#5819)
---
 .../tx/impl/TransactionExpirationRegistry.java     | 28 +++++++++++++++++-----
 1 file changed, 22 insertions(+), 6 deletions(-)

diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
index afd20b579b2..5dec6555841 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
@@ -18,10 +18,11 @@
 package org.apache.ignite.internal.tx.impl;
 
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
 import java.util.NavigableMap;
+import java.util.NoSuchElementException;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import org.apache.ignite.internal.hlc.HybridTimestamp;
 import org.apache.ignite.internal.logger.IgniteLogger;
@@ -91,7 +92,10 @@ class TransactionExpirationRegistry {
                         if (txOrSet instanceof Set) {
                             txsExpiringAtTs = (Set<InternalTransaction>) txOrSet;
                         } else {
-                            txsExpiringAtTs = new HashSet<>();
+                            // Using a concurrent set because txsByExpirationTime does not guarantee linearization of compute() calls,
+                            // so the set might be accessed/updated concurrently. It's not a problem to use a mutable set here as both
+                            // addition to it and removal from it are idempotent.
+                            txsExpiringAtTs = ConcurrentHashMap.newKeySet();
                             txsExpiringAtTs.add((InternalTransaction) txOrSet);
                         }
 
@@ -160,15 +164,27 @@ class TransactionExpirationRegistry {
 
                 set.remove(tx);
 
-                return set.size() == 1 ? set.iterator().next() : set;
+                int newSize = set.size();
+
+                if (newSize == 0) {
+                    return null;
+                } else if (newSize == 1) {
+                    try {
+                        return set.iterator().next();
+                    } catch (NoSuchElementException e) {
+                        return null;
+                    }
+                } else {
+                    return set;
+                }
             } else {
-                InternalTransaction tx0 = (InternalTransaction) txOrSet;
+                InternalTransaction registeredTx = (InternalTransaction) txOrSet;
 
-                if (tx0.id().equals(tx.id())) {
+                if (registeredTx.id().equals(tx.id())) {
                     return null;
                 }
 
-                return tx0;
+                return registeredTx;
             }
         });
     }