You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "ptupitsyn (via GitHub)" <gi...@apache.org> on 2026/03/13 12:23:49 UTC
[PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
ptupitsyn opened a new pull request, #7779:
URL: https://github.com/apache/ignite-3/pull/7779
(no comment)
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779#discussion_r2945978507
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionCleanupTest.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.runner.app.client;
+
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.runner.app.client.ItThinClientTransactionsTest.generateKeysForNode;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.partition.Partition;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for client transaction cleanup on disconnect.
+ */
+@SuppressWarnings({"resource", "DataFlowIssue"})
+public class ItThinClientTransactionCleanupTest extends ItAbstractThinClientTest {
+ /**
+ * Tests that locks are released when client disconnects with a transaction having direct enlistments.
+ */
+ @Test
+ void testClientDisconnectReleasesTxLocksFast() {
+ try (IgniteClient client = IgniteClient.builder().addresses(getClientAddresses().toArray(new String[0])).build()) {
+ var table = (ClientTable) client.tables().table(TABLE_NAME);
+ Map<Partition, ClusterNode> map = table.partitionDistribution().primaryReplicas();
+
+ IgniteImpl server0 = unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(300, 1, map, server0.cluster().localNode(), table);
+ List<Tuple> tuples1 = generateKeysForNode(310, 1, map, server1.cluster().localNode(), table);
+
+ // Perform a mix of operations to trigger direct tx logic.
+ Map<Tuple, Tuple> data = new HashMap<>();
+
+ data.put(tuples0.get(0), val(tuples0.get(0).intValue(0) + ""));
+ data.put(tuples1.get(0), val(tuples1.get(0).intValue(0) + ""));
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client.transactions().begin();
+
+ table.keyValueView().putAll(tx0, data);
+
+ for (Entry<Tuple, Tuple> entry : data.entrySet()) {
+ table.keyValueView().put(tx0, entry.getKey(), entry.getValue());
+ }
+
+ assertThat(txLockCount(), greaterThanOrEqualTo(2));
+
+ // Disconnect without commit or rollback.
+ }
+
+ await().atMost(Duration.ofSeconds(2))
+ .untilAsserted(() -> assertEquals(0, txLockCount()));
+ }
+
+ private int txLockCount() {
+ int count = 0;
+
+ for (int i = 0; i < nodes(); i++) {
+ IgniteImpl ignite = unwrapIgniteImpl(server(i));
+ LockManager lockManager = ignite.txManager().lockManager();
+
+ var iter = lockManager.locks();
Review Comment:
Fixed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn merged PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779#discussion_r2930944032
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java:
##########
@@ -138,6 +182,18 @@ public void close() {
res.clear();
+ for (var cleaner : txCleaners.values()) {
+ try {
+ cleaner.clean().join();
+ } catch (Throwable e) {
+ if (ex == null) {
+ ex = new IgniteInternalException(e);
+ } else {
+ ex.addSuppressed(e);
+ }
+ }
+ }
Review Comment:
`ClientResourceRegistry.close()` is called from Netty `channelInactive` (event loop thread), but it blocks by doing `cleaner.clean().join()` for every tracked tx. This can stall the event loop during disconnect storms or slow cleanup, impacting other connections and potentially deadlocking if cleanup completion depends on the same event loop. Prefer non-blocking cleanup (fire-and-forget with logging), or aggregate futures and run/await them on a separate executor with a timeout.
##########
modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java:
##########
@@ -295,6 +296,11 @@ public int lockRetryCount() {
return 0;
}
+ @Override
+ public RemotelyTriggeredResourceRegistry resourceRegistry() {
+ return null;
+ }
Review Comment:
`TxManager.resourceRegistry()` is now part of the interface and is used by server-side code (e.g., direct-tx cleanup registration). Returning `null` here can lead to NPEs if any test paths exercise that logic. Provide a non-null stub implementation (for example, a new `RemotelyTriggeredResourceRegistry`) or throw `UnsupportedOperationException` if the fake is not meant to be used in such scenarios.
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java:
##########
@@ -538,6 +539,24 @@ public static TableNotFoundException tableIdNotFoundException(Integer tableId) {
MESSAGE_TX_ALREADY_FINISHED_DUE_TO_TIMEOUT + " [tx=" + remote + "].");
}
+ // Track this remote enlistment for cleanup if client disconnects.
+ try {
+ resources.addTxCleaner(txId, tableId, commitPart, txManager, (IgniteTablesInternal) tables);
Review Comment:
`commitPart` here is the transaction *commit* partition sent by the client, not the partition that this request actually operates on. Recording enlistments using `commitPart` will make cleanup target the wrong replication group on most direct-mapping operations (multi-partition tx), and can even fail cleanup on disconnect if the local node doesn't host that partition. The cleaner should record the *actual* partition/group that was enlisted on this node (e.g., derive it from the current request's target replication group / computed partition from the key(s)), not the commit partition.
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTxPartitionEnlistmentCleaner.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.client.handler.requests.tx;
+
+import static java.util.stream.Collectors.toList;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
+
+/**
+ * Helper class to clean up direct transaction enlistments on the client side.
Review Comment:
The class-level Javadoc says "clean up ... on the client side", but this helper lives in `client-handler` and runs on the server to clean up server-side state for a client connection. Update the wording to avoid confusion when maintaining the cleanup flow.
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTxPartitionEnlistmentCleaner.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.client.handler.requests.tx;
+
+import static java.util.stream.Collectors.toList;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
+
+/**
+ * Helper class to clean up direct transaction enlistments on the client side.
+ */
+public class ClientTxPartitionEnlistmentCleaner {
+ private final UUID txId;
+
+ private final TxManager txManager;
+
+ private final IgniteTablesInternal igniteTables;
+
+ private final Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedPartitions = new HashMap<>();
+
+ /**
+ * Creates a new instance of the transaction partition enlistment cleaner.
+ *
+ * @param txId Transaction ID.
+ * @param txManager Transaction manager.
+ * @param igniteTables Ignite tables.
+ */
+ public ClientTxPartitionEnlistmentCleaner(UUID txId, TxManager txManager, IgniteTablesInternal igniteTables) {
+ this.txId = txId;
+ this.txManager = txManager;
+ this.igniteTables = igniteTables;
+ }
+
+ /**
+ * Adds a partition enlistment for the given table and partition.
+ *
+ * @param tableId Table ID.
+ * @param partId Partition ID.
+ */
+ public void addEnlistment(int tableId, int partId) {
+ TableViewInternal table = igniteTables.cachedTable(tableId);
+
+ if (table != null) {
+ ZonePartitionId replicationGroupId = table.internalTable().targetReplicationGroupId(partId);
+ enlistedPartitions.computeIfAbsent(replicationGroupId, k -> new PendingTxPartitionEnlistment(null, 0))
+ .addTableId(tableId);
+ }
Review Comment:
`ClientTxPartitionEnlistmentCleaner` is stored per-connection and can be updated from multiple concurrent partition-operation threads (client requests are executed on `partitionOperationsExecutor`). However `enlistedPartitions` is a plain `HashMap` with non-atomic updates, so concurrent `addEnlistment` calls can race and corrupt state. Make this structure thread-safe (e.g., use `ConcurrentHashMap` and concurrent sets for tableIds, or synchronize `addEnlistment`/`clean`) to avoid lost/invalid enlistments and cleanup failures.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionCleanupTest.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.runner.app.client;
+
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.runner.app.client.ItThinClientTransactionsTest.generateKeysForNode;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.partition.Partition;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for client transaction cleanup on disconnect.
+ */
+@SuppressWarnings({"resource", "DataFlowIssue"})
+public class ItThinClientTransactionCleanupTest extends ItAbstractThinClientTest {
+ /**
+ * Tests that locks are released when client disconnects with a transaction having direct enlistments.
+ */
+ @Test
+ void testClientDisconnectReleasesTxLocksFast() {
+ try (IgniteClient client = IgniteClient.builder().addresses(getClientAddresses().toArray(new String[0])).build()) {
+ var table = (ClientTable) client.tables().table(TABLE_NAME);
+ Map<Partition, ClusterNode> map = table.partitionDistribution().primaryReplicas();
+
+ IgniteImpl server0 = unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(300, 1, map, server0.cluster().localNode(), table);
+ List<Tuple> tuples1 = generateKeysForNode(310, 1, map, server1.cluster().localNode(), table);
+
+ // Perform a mix of operations to trigger direct tx logic.
+ Map<Tuple, Tuple> data = new HashMap<>();
+
+ data.put(tuples0.get(0), val(tuples0.get(0).intValue(0) + ""));
+ data.put(tuples1.get(0), val(tuples1.get(0).intValue(0) + ""));
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client().transactions().begin();
Review Comment:
The test opens a new IgniteClient in try-with-resources, but starts the transaction via `client().transactions().begin()` (the shared client from the base class) instead of the local `client`. Closing the local client then doesn't necessarily disconnect the transaction being used by the operations, so the test can pass/fail for the wrong reason. Use the locally created `client` to begin the transaction (and to perform the operations) so the disconnect being tested matches the transaction under test.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779#discussion_r2931194386
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java:
##########
@@ -538,6 +539,24 @@ public static TableNotFoundException tableIdNotFoundException(Integer tableId) {
MESSAGE_TX_ALREADY_FINISHED_DUE_TO_TIMEOUT + " [tx=" + remote + "].");
}
+ // Track this remote enlistment for cleanup if client disconnects.
+ try {
+ resources.addTxCleaner(txId, tableId, commitPart, txManager, (IgniteTablesInternal) tables);
+ } catch (IgniteInternalCheckedException e) {
+ // Client disconnected (resource registry closed).
+ try {
+ remote.rollback();
+ } catch (Exception ex) {
+ e.addSuppressed(ex);
+ }
+
+ throw new IgniteException(e.traceId(), e.code(), "Client disconnected, tx rolled back: " + remote, e);
+ }
+
+ // Stop tracking on tx finish.
+ txManager.resourceRegistry().register(
+ new FullyQualifiedResourceId(txId, txId), txId, () -> () -> resources.removeTxCleaner(txId));
Review Comment:
The second argument to `RemotelyTriggeredResourceRegistry.register` is `remoteHostId` (creator node ID). Passing `txId` here will make the resource vacuum treat every tracked tx as “orphaned” (since `clusterNodeResolver.getById(txId)` will always be null) and close it prematurely, removing the cleaner while the transaction is still active. Use a real cluster node id for `remoteHostId` (e.g., the coordinator id `coord` from the request) so orphan detection doesn’t unregister the cleaner immediately.
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTxPartitionEnlistmentCleaner.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.client.handler.requests.tx;
+
+import static java.util.stream.Collectors.toList;
+
+import java.util.HashMap;
Review Comment:
`java.util.HashMap` is imported but never used in this class. Unused imports fail compilation under typical build settings (e.g., with `-Werror` / Checkstyle), so this should be removed.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "ascherbakoff (via GitHub)" <gi...@apache.org>.
ascherbakoff commented on code in PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779#discussion_r2945890867
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTxPartitionEnlistmentCleaner.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.client.handler.requests.tx;
+
+import static java.util.stream.Collectors.toList;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
+
+/**
+ * Helper class to clean up direct transaction enlistments on the server side for a client connection.
+ */
+public class ClientTxPartitionEnlistmentCleaner {
+ private final UUID txId;
+
+ private final TxManager txManager;
+
+ private final IgniteTablesInternal igniteTables;
+
+ private final Map<ZonePartitionId, PendingTxPartitionEnlistment> enlistedPartitions = new ConcurrentHashMap<>();
+
+ /**
+ * Creates a new instance of the transaction partition enlistment cleaner.
+ *
+ * @param txId Transaction ID.
+ * @param txManager Transaction manager.
+ * @param igniteTables Ignite tables.
+ */
+ public ClientTxPartitionEnlistmentCleaner(UUID txId, TxManager txManager, IgniteTablesInternal igniteTables) {
+ this.txId = txId;
+ this.txManager = txManager;
+ this.igniteTables = igniteTables;
+ }
+
+ /**
+ * Adds a partition enlistment for the given table and partition.
+ *
+ * @param tableId Table ID.
+ * @param partId Partition ID.
+ */
+ public void addEnlistment(int tableId, int partId) {
+ TableViewInternal table = igniteTables.cachedTable(tableId);
+
+ if (table != null) {
+ ZonePartitionId replicationGroupId = table.internalTable().targetReplicationGroupId(partId);
+ enlistedPartitions.computeIfAbsent(replicationGroupId, k -> new PendingTxPartitionEnlistment(null, 0))
+ .addTableId(tableId);
+ }
+ }
+
+ /**
+ * Discards local write intents for all enlisted partitions.
+ *
+ * @return Future that completes when cleanup is done.
+ */
+ public CompletableFuture<Void> clean() {
+ List<EnlistedPartitionGroup> enlistedPartitionGroups = enlistedPartitions.entrySet().stream()
Review Comment:
I believe it is not safe to simply drop the transaction part on client's disconnect.
This means locks can be dropped while the transaction is still active, causing data corruption.
We need first to mark the transaction as aborted on a commit partition.
@denis-chudov Can you provide the required method to achieve this ?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionCleanupTest.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.runner.app.client;
+
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.runner.app.client.ItThinClientTransactionsTest.generateKeysForNode;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.partition.Partition;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for client transaction cleanup on disconnect.
+ */
+@SuppressWarnings({"resource", "DataFlowIssue"})
+public class ItThinClientTransactionCleanupTest extends ItAbstractThinClientTest {
+ /**
+ * Tests that locks are released when client disconnects with a transaction having direct enlistments.
+ */
+ @Test
+ void testClientDisconnectReleasesTxLocksFast() {
+ try (IgniteClient client = IgniteClient.builder().addresses(getClientAddresses().toArray(new String[0])).build()) {
+ var table = (ClientTable) client.tables().table(TABLE_NAME);
+ Map<Partition, ClusterNode> map = table.partitionDistribution().primaryReplicas();
+
+ IgniteImpl server0 = unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(300, 1, map, server0.cluster().localNode(), table);
+ List<Tuple> tuples1 = generateKeysForNode(310, 1, map, server1.cluster().localNode(), table);
+
+ // Perform a mix of operations to trigger direct tx logic.
+ Map<Tuple, Tuple> data = new HashMap<>();
+
+ data.put(tuples0.get(0), val(tuples0.get(0).intValue(0) + ""));
+ data.put(tuples1.get(0), val(tuples1.get(0).intValue(0) + ""));
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client.transactions().begin();
+
+ table.keyValueView().putAll(tx0, data);
+
+ for (Entry<Tuple, Tuple> entry : data.entrySet()) {
+ table.keyValueView().put(tx0, entry.getKey(), entry.getValue());
+ }
+
+ assertThat(txLockCount(), greaterThanOrEqualTo(2));
+
+ // Disconnect without commit or rollback.
+ }
+
+ await().atMost(Duration.ofSeconds(2))
+ .untilAsserted(() -> assertEquals(0, txLockCount()));
+ }
+
+ private int txLockCount() {
+ int count = 0;
+
+ for (int i = 0; i < nodes(); i++) {
+ IgniteImpl ignite = unwrapIgniteImpl(server(i));
+ LockManager lockManager = ignite.txManager().lockManager();
+
+ var iter = lockManager.locks();
Review Comment:
You can use `CollectionUtils.count(locks)`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-27651 Clean up direct tx on client disconnect [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7779:
URL: https://github.com/apache/ignite-3/pull/7779#discussion_r2938319331
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java:
##########
@@ -538,6 +539,24 @@ public static TableNotFoundException tableIdNotFoundException(Integer tableId) {
MESSAGE_TX_ALREADY_FINISHED_DUE_TO_TIMEOUT + " [tx=" + remote + "].");
}
+ // Track this remote enlistment for cleanup if client disconnects.
+ try {
+ resources.addTxCleaner(txId, tableId, commitPart, txManager, (IgniteTablesInternal) tables);
+ } catch (IgniteInternalCheckedException e) {
+ // Client disconnected (resource registry closed).
+ try {
+ remote.rollback();
+ } catch (Exception ex) {
+ e.addSuppressed(ex);
+ }
+
+ throw new IgniteException(e.traceId(), e.code(), "Client disconnected, tx rolled back: " + remote, e);
+ }
+
+ // Stop tracking on tx finish.
+ txManager.resourceRegistry().register(
+ new FullyQualifiedResourceId(txId, txId), txId, () -> () -> resources.removeTxCleaner(txId));
Review Comment:
In this register() call, the second argument (`remoteHostId`) is supposed to be the cluster node UUID that created/triggered the resource. Passing `txId` here will make ResourceVacuumManager treat it as an orphan (no such node in topology) and close it on the next vacuum cycle, which can remove the cleaner while the tx is still active and break disconnect cleanup. Use the coordinator node id (`coord`) or another real cluster node id as `remoteHostId` instead of `txId`.
##########
modules/core/src/main/java/org/apache/ignite/internal/util/ExceptionUtils.java:
##########
@@ -455,6 +456,31 @@ private static boolean hasCauseOrSuppressedInternal(
return false;
}
+ /**
+ * Checks if the given throwable is already present in the cause or suppressed hierarchy of the given throwable.
+ *
+ * @param t Throwable.
+ * @param dejaVu Known throwables.
+ * @return True if seen before, false otherwise.
+ */
+ public static boolean existingCauseOrSuppressed(Throwable t, HashSet<Throwable> dejaVu) {
+ if (t == null) {
+ return false;
+ }
+
+ if (!dejaVu.add(t)) {
+ return true;
+ }
+
+ for (Throwable sup : t.getSuppressed()) {
+ if (existingCauseOrSuppressed(sup, dejaVu)) {
+ return true;
+ }
+ }
+
+ return existingCauseOrSuppressed(t.getCause(), dejaVu);
+ }
Review Comment:
The new `existingCauseOrSuppressed` helper exposes `HashSet<Throwable>` in its public signature and even accepts `t == null` while the parameter isn't marked nullable. Consider changing the signature to use `Set<Throwable>` (and annotate `t` as `@Nullable` if null is supported) so callers can reuse the IdentityHashMap-backed Set pattern already used elsewhere in this class and so the API isn't tied to a concrete collection type.
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java:
##########
@@ -112,6 +123,44 @@ public ClientResource remove(long id) throws IgniteInternalCheckedException {
}
}
+ /**
+ * Records that a remote transaction enlisted a partition on this node.
+ *
+ * @param txId Transaction ID.
+ * @param tableId Table ID.
+ * @param partitionId Partition ID.
Review Comment:
The Javadoc for `addTxCleaner` doesn't mention the `txManager` and `tables` parameters that are part of the method signature. Please update the comment so it documents all parameters (or adjust the signature if those should be stored in the registry instead).
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientResourceRegistry.java:
##########
@@ -122,24 +171,35 @@ public void close() {
busyLock.block();
- IgniteInternalException ex = null;
+ AtomicReference<IgniteInternalException> ex = new AtomicReference<>();
+ var dejaVu = new HashSet<Throwable>();
- for (ClientResource r : res.values()) {
+ Consumer<Runnable> releaseSafe = r -> {
try {
- r.release();
- } catch (Exception e) {
- if (ex == null) {
- ex = new IgniteInternalException(e);
- } else {
- ex.addSuppressed(e);
+ r.run();
+ } catch (Throwable e) {
+ if (ex.get() == null) {
+ ex.set(new IgniteInternalException(e));
+ existingCauseOrSuppressed(e, dejaVu); // Seed dejaVu.
+ } else if (!existingCauseOrSuppressed(e, dejaVu)) {
+ ex.get().addSuppressed(e);
}
}
+ };
+
+ for (ClientResource r : res.values()) {
+ releaseSafe.accept(r::release);
}
res.clear();
- if (ex != null) {
- throw ex;
+ for (var cleaner : txCleaners.values()) {
+ // Don't block the thread, clean in background. discardLocalWriteIntents swallows errors anyway.
+ releaseSafe.accept(cleaner::clean);
Review Comment:
`releaseSafe` accepts a `Runnable`, but `ClientTxPartitionEnlistmentCleaner.clean()` returns `CompletableFuture<Void>`, so `cleaner::clean` is not compatible with `Runnable` and should not compile. Wrap it in a void-compatible lambda (e.g., invoke `clean()` inside `Runnable`) or change `releaseSafe` to accept a functional type that can return a value (Supplier/Callable).
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org