This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 c9a6a3af1dc IGNITE-26567 Fix leftover transaction contexts on
non-coordinator nodes after RO tx completion (#7649)
c9a6a3af1dc is described below
commit c9a6a3af1dcbd55a7955c962e65fb43d93051aa6
Author: Anton Laletin <[email protected]>
AuthorDate: Thu Mar 5 17:48:22 2026 +0400
IGNITE-26567 Fix leftover transaction contexts on non-coordinator nodes
after RO tx completion (#7649)
---
.../internal/runner/app/client/ItThinClientSqlTest.java | 13 ++++++++++---
.../tx/impl/FinishedTransactionBatchRequestHandler.java | 8 ++++++++
.../ignite/internal/tx/impl/ResourceVacuumManager.java | 1 +
.../tx/impl/FinishedTransactionBatchRequestHandlerTest.java | 6 ++++++
4 files changed, 25 insertions(+), 3 deletions(-)
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
index 60b14cb3ba9..0b2a21923c6 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.runner.app.client;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
+import static
org.apache.ignite.internal.tx.impl.ResourceVacuumManager.RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
@@ -31,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.math.BigDecimal;
+import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -38,6 +40,7 @@ import java.time.ZoneId;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletionException;
@@ -51,6 +54,7 @@ import org.apache.ignite.internal.client.sql.ClientSql;
import org.apache.ignite.internal.client.sql.QueryModifier;
import org.apache.ignite.internal.security.authentication.UserDetails;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.lang.IgniteException;
@@ -73,7 +77,6 @@ import org.apache.ignite.tx.TransactionOptions;
import org.awaitility.Awaitility;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -81,6 +84,7 @@ import org.junit.jupiter.params.provider.ValueSource;
/**
* Thin client SQL integration test.
*/
+@WithSystemProperty(key = RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY,
value = "10")
@SuppressWarnings("resource")
public class ItThinClientSqlTest extends ItAbstractThinClientTest {
@AfterEach
@@ -840,7 +844,6 @@ public class ItThinClientSqlTest extends
ItAbstractThinClientTest {
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-26567";)
public void testBroadcastQueryTxInflightStateCleanup() {
IgniteSql sql = client().sql();
@@ -858,7 +861,11 @@ public class ItThinClientSqlTest extends
ItAbstractThinClientTest {
IgniteImpl server = TestWrappers.unwrapIgniteImpl(server(i));
TxManager txManager = server.txManager();
TransactionInflights transactionInflights =
IgniteTestUtils.getFieldValue(txManager, "transactionInflights");
- assertFalse(transactionInflights.hasActiveInflights(), "Expecting
no active inflights");
+ Map<UUID, ?> txContexts =
IgniteTestUtils.getFieldValue(transactionInflights, "txCtxMap");
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(10))
+ .untilAsserted(() -> assertTrue(txContexts.isEmpty(),
"Expecting no transaction inflight contexts"));
}
}
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandler.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandler.java
index 6e265624939..171707c098c 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandler.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandler.java
@@ -34,6 +34,9 @@ public class FinishedTransactionBatchRequestHandler {
/** Resources registry. */
private final RemotelyTriggeredResourceRegistry resourcesRegistry;
+ /** Transaction inflights tracker. */
+ private final TransactionInflights transactionInflights;
+
private final LowWatermark lowWatermark;
private final Executor asyncExecutor;
@@ -43,17 +46,20 @@ public class FinishedTransactionBatchRequestHandler {
*
* @param messagingService Messaging service.
* @param resourcesRegistry Resources registry.
+ * @param transactionInflights Transaction inflights.
* @param lowWatermark Low watermark.
* @param asyncExecutor Executor to run cleanup commands.
*/
public FinishedTransactionBatchRequestHandler(
MessagingService messagingService,
RemotelyTriggeredResourceRegistry resourcesRegistry,
+ TransactionInflights transactionInflights,
LowWatermark lowWatermark,
Executor asyncExecutor
) {
this.messagingService = messagingService;
this.resourcesRegistry = resourcesRegistry;
+ this.transactionInflights = transactionInflights;
this.lowWatermark = lowWatermark;
this.asyncExecutor = asyncExecutor;
}
@@ -74,6 +80,8 @@ public class FinishedTransactionBatchRequestHandler {
}
private void cleanUpForTransaction(UUID transactionId) {
+ transactionInflights.removeTxContext(transactionId);
+
resourcesRegistry.close(transactionId);
lowWatermark.unlock(transactionId);
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ResourceVacuumManager.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ResourceVacuumManager.java
index a76cc3fa43a..bcbaaf9fc0b 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ResourceVacuumManager.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ResourceVacuumManager.java
@@ -124,6 +124,7 @@ public class ResourceVacuumManager implements
IgniteComponent {
this.finishedTransactionBatchRequestHandler = new
FinishedTransactionBatchRequestHandler(
messagingService,
resourceRegistry,
+ transactionInflights,
lowWatermark,
resourceVacuumExecutor
);
diff --git
a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandlerTest.java
b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandlerTest.java
index ba1ba7cc25a..e879d0b6b83 100644
---
a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandlerTest.java
+++
b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/FinishedTransactionBatchRequestHandlerTest.java
@@ -54,6 +54,9 @@ class FinishedTransactionBatchRequestHandlerTest extends
BaseIgniteAbstractTest
@Mock
private LowWatermark lowWatermark;
+ @Mock
+ private TransactionInflights transactionInflights;
+
private FinishedTransactionBatchRequestHandler requestHandler;
private NetworkMessageHandler networkHandler;
@@ -63,6 +66,7 @@ class FinishedTransactionBatchRequestHandlerTest extends
BaseIgniteAbstractTest
requestHandler = new FinishedTransactionBatchRequestHandler(
messagingService,
resourceRegistry,
+ transactionInflights,
lowWatermark,
ForkJoinPool.commonPool()
);
@@ -86,6 +90,8 @@ class FinishedTransactionBatchRequestHandlerTest extends
BaseIgniteAbstractTest
networkHandler.onReceived(message, mock(InternalClusterNode.class),
null);
+ verify(transactionInflights, timeout(10_000)).removeTxContext(txId1);
+ verify(transactionInflights, timeout(10_000)).removeTxContext(txId2);
verify(lowWatermark, timeout(10_000)).unlock(txId1);
verify(lowWatermark, timeout(10_000)).unlock(txId2);
}
