You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2025/02/20 14:45:48 UTC
(ignite-3) branch main updated: IGNITE-24366 Move VacuumTxStateReplicaRequest processing to zone replica listener (#5245)
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 cd8e58e9fae IGNITE-24366 Move VacuumTxStateReplicaRequest processing to zone replica listener (#5245)
cd8e58e9fae is described below
commit cd8e58e9faecee7eeb17e8311a8e7433505f2d0a
Author: Slava Koptilin <sl...@gmail.com>
AuthorDate: Thu Feb 20 16:45:41 2025 +0200
IGNITE-24366 Move VacuumTxStateReplicaRequest processing to zone replica listener (#5245)
---
.../descriptors/CatalogTableDescriptor.java | 2 +-
.../internal/lang/IgniteSystemProperties.java | 2 +-
.../replicator/ItAbstractColocationTest.java | 172 ++++++++-------
.../replicator/ItCatalogCompactionTest.java | 181 ++++++++++++++++
.../replicator/ItKeyValueViewOperationsTest.java | 7 +-
.../replicator/ItReplicaLifecycleTest.java | 231 +++++----------------
.../replicator/ItTransactionsVacuumTest.java | 142 +++++++++++++
.../partition/replicator/fixtures/Node.java | 21 +-
.../ReplicationRaftCommandApplicator.java | 17 +-
.../replicator/ZonePartitionReplicaListener.java | 10 +-
.../MinimumActiveTxTimeReplicaRequestHandler.java | 2 +-
.../handlers/TxFinishReplicaRequestHandler.java | 2 +-
.../VacuumTxStateReplicaRequestHandler.java | 59 ++++++
.../replicator/raft/ZonePartitionRaftListener.java | 65 ++++--
.../handlers/VacuumTxStatesCommandHandler.java | 59 ++++++
.../replicator/message/ReplicaMessageUtils.java | 5 +-
.../benchmark/AbstractColocationBenchmark.java | 2 +-
.../disaster/PartitionStatesMetricSource.java | 2 +-
.../table/distributed/raft/PartitionListener.java | 40 ++--
.../MinimumActiveTxTimeCommandHandler.java | 6 +-
.../replicator/PartitionReplicaListener.java | 23 +-
.../tx/impl/PersistentTxStateVacuumizer.java | 19 +-
.../ignite/internal/tx/impl/TxManagerImpl.java | 4 +-
.../tx/impl/VolatileTxStateMetaStorage.java | 12 +-
24 files changed, 745 insertions(+), 340 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptor.java
index e035380494d..989f3ed9c68 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptor.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogTableDescriptor.java
@@ -267,7 +267,7 @@ public class CatalogTableDescriptor extends CatalogObjectDescriptor {
}
/**
- * Returns {@code true} if this the given column is a part of collocation key.
+ * Returns {@code true} if this the given column is a part of colocation key.
*/
public boolean isColocationColumn(String name) {
return colocationColumns.contains(name);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/lang/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/internal/lang/IgniteSystemProperties.java
index 8c2029c9159..c53d244498f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/lang/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/lang/IgniteSystemProperties.java
@@ -91,7 +91,7 @@ public final class IgniteSystemProperties {
}
// TODO https://issues.apache.org/jira/browse/IGNITE-22522 Remove.
- /* Feature flag for zone based collocation track */
+ /* Feature flag for zone based colocation track */
public static boolean enabledColocation() {
return getBoolean(COLOCATION_FEATURE_FLAG, false);
}
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItAbstractColocationTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItAbstractColocationTest.java
index 1ddab6d51be..a1b54f5bb09 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItAbstractColocationTest.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItAbstractColocationTest.java
@@ -29,11 +29,11 @@ import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.apache.ignite.sql.ColumnType.INT32;
import static org.apache.ignite.sql.ColumnType.INT64;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import java.util.HashMap;
+import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.IntStream;
@@ -45,8 +45,8 @@ import org.apache.ignite.internal.configuration.testframework.ConfigurationExten
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
+import org.apache.ignite.internal.network.NodeFinder;
import org.apache.ignite.internal.network.StaticNodeFinder;
import org.apache.ignite.internal.partition.replicator.fixtures.Node;
import org.apache.ignite.internal.partition.replicator.fixtures.TestPlacementDriver;
@@ -68,107 +68,115 @@ import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
+// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
+
+/**
+ * Base class for tests that require a cluster with zone replication.
+ */
@ExtendWith(ConfigurationExtension.class)
-@ExtendWith(ExecutorServiceExtension.class)
@ExtendWith(SystemPropertiesExtension.class)
-// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
+@ExtendWith(ExecutorServiceExtension.class)
@WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
abstract class ItAbstractColocationTest extends IgniteAbstractTest {
private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
- private static final int NODE_COUNT = 3;
-
private static final int BASE_PORT = 20_000;
- private static final String HOST = "localhost";
-
- private static final List<NetworkAddress> NODE_ADDRESSES = IntStream.range(0, NODE_COUNT)
- .mapToObj(i -> new NetworkAddress(HOST, BASE_PORT + i))
- .collect(toList());
-
- private static final StaticNodeFinder NODE_FINDER = new StaticNodeFinder(NODE_ADDRESSES);
-
static final int AWAIT_TIMEOUT_MILLIS = 10_000;
- @InjectConfiguration("mock.nodeAttributes: {region = US, storage = SSD}")
- private static NodeAttributesConfiguration nodeAttributes1;
-
- @InjectConfiguration("mock.nodeAttributes: {region = EU, storage = SSD}")
- private static NodeAttributesConfiguration nodeAttributes2;
-
- @InjectConfiguration("mock.nodeAttributes: {region = UK, storage = SSD}")
- private static NodeAttributesConfiguration nodeAttributes3;
-
@InjectConfiguration
- private static TransactionConfiguration txConfiguration;
+ private SystemLocalConfiguration systemConfiguration;
@InjectConfiguration
- private static RaftConfiguration raftConfiguration;
+ private RaftConfiguration raftConfiguration;
@InjectConfiguration
- private static SystemLocalConfiguration systemConfiguration;
+ private NodeAttributesConfiguration defaultNodeAttributesConfiguration;
+
+ @InjectConfiguration("mock.profiles = {" + DEFAULT_STORAGE_PROFILE + ".engine = aipersist, test.engine=test}")
+ private StorageConfiguration storageConfiguration;
@InjectConfiguration
- private static ReplicationConfiguration replicationConfiguration;
+ private MetaStorageConfiguration metaStorageConfiguration;
@InjectConfiguration
- private static MetaStorageConfiguration metaStorageConfiguration;
+ private ReplicationConfiguration replicationConfiguration;
- @InjectConfiguration("mock.profiles = {" + DEFAULT_STORAGE_PROFILE + ".engine = aipersist, test.engine=test}")
- private static StorageConfiguration storageConfiguration;
+ @InjectExecutorService
+ private ScheduledExecutorService scheduledExecutorService;
@InjectConfiguration
GcConfiguration gcConfiguration;
- @InjectExecutorService
- private static ScheduledExecutorService scheduledExecutorService;
-
- private static List<NodeAttributesConfiguration> nodeAttributesConfigurations;
+ @InjectConfiguration
+ TransactionConfiguration txConfiguration;
- final Map<Integer, Node> nodes = new HashMap<>();
+ final List<Node> cluster = new ArrayList<>();
// TODO Remove https://issues.apache.org/jira/browse/IGNITE-24375
final TestPlacementDriver placementDriver = new TestPlacementDriver();
- @BeforeAll
- static void beforeAll() {
- nodeAttributesConfigurations = List.of(nodeAttributes1, nodeAttributes2, nodeAttributes3);
+ private NodeFinder nodeFinder;
+
+ private TestInfo testInfo;
+
+ @BeforeEach
+ void setUp(TestInfo testInfo) {
+ this.testInfo = testInfo;
}
@AfterEach
- void after() throws Exception {
- closeAll(nodes.values().parallelStream().map(node -> node::stop));
+ void tearDown() throws Exception {
+ closeAll(cluster.parallelStream().map(node -> node::stop));
+ }
+
+ void startCluster(int size) throws Exception {
+ startCluster(size, null, null);
}
- void startNodes(TestInfo testInfo, int amount) throws NodeStoppingException, InterruptedException {
- startNodes(testInfo, amount, null);
+ void startCluster(int size, List<NodeAttributesConfiguration> customAttributes) throws Exception {
+ assertThat(customAttributes.size(), equalTo(size));
+
+ startCluster(size, null, customAttributes);
}
- void startNodes(
- TestInfo testInfo,
- int amount,
- @Nullable Node.InvokeInterceptor invokeInterceptor
- ) throws NodeStoppingException, InterruptedException {
- IntStream.range(0, amount).forEach(i -> newNode(testInfo, i, invokeInterceptor));
+ void startCluster(
+ int size,
+ @Nullable Node.InvokeInterceptor invokeInterceptor,
+ @Nullable List<NodeAttributesConfiguration> customAttributes
+ ) throws Exception {
+ List<NetworkAddress> addresses = IntStream.range(0, size)
+ .mapToObj(i -> new NetworkAddress("localhost", BASE_PORT + i))
+ .collect(toList());
+
+ nodeFinder = new StaticNodeFinder(addresses);
- assert nodes.size() == amount : "Not all amount of nodes were created.";
+ boolean hasCustomAttributes = customAttributes != null;
- nodes.values().stream().parallel().forEach(Node::start);
+ IntStream.range(0, size)
+ .mapToObj(i -> newNode(
+ addresses.get(i),
+ nodeFinder,
+ invokeInterceptor,
+ hasCustomAttributes ? customAttributes.get(i) : defaultNodeAttributesConfiguration))
+ .forEach(cluster::add);
- Node node0 = getNode(0);
+ cluster.parallelStream().forEach(Node::start);
+
+ Node node0 = cluster.get(0);
node0.cmgManager.initCluster(List.of(node0.name), List.of(node0.name), "cluster");
placementDriver.setPrimary(node0.clusterService.topologyService().localMember());
- nodes.values().forEach(Node::waitWatches);
+ cluster.forEach(Node::waitWatches);
assertThat(
- allOf(nodes.values().stream().map(n -> n.cmgManager.onJoinReady()).toArray(CompletableFuture[]::new)),
+ allOf(cluster.stream().map(n -> n.cmgManager.onJoinReady()).toArray(CompletableFuture[]::new)),
willCompleteSuccessfully()
);
@@ -178,19 +186,20 @@ abstract class ItAbstractColocationTest extends IgniteAbstractTest {
assertThat(logicalTopologyFuture, willCompleteSuccessfully());
- return logicalTopologyFuture.join().nodes().size() == amount;
+ return logicalTopologyFuture.join().nodes().size() == cluster.size();
},
AWAIT_TIMEOUT_MILLIS
));
}
-
- Node startNode(TestInfo testInfo, int idx) {
- return startNode(testInfo, idx, null);
+ protected Node addNodeToCluster() {
+ return addNodeToCluster(cluster.size());
}
- private Node startNode(TestInfo testInfo, int idx, @Nullable Node.InvokeInterceptor invokeInterceptor) {
- Node node = newNode(testInfo, idx, invokeInterceptor);
+ Node addNodeToCluster(int idx) {
+ Node node = newNode(new NetworkAddress("localhost", BASE_PORT + idx), nodeFinder);
+
+ cluster.add(node);
node.start();
@@ -206,19 +215,28 @@ abstract class ItAbstractColocationTest extends IgniteAbstractTest {
node.stop();
- nodes.remove(idx);
+ cluster.remove(idx);
}
- private Node newNode(TestInfo testInfo, int idx, @Nullable Node.InvokeInterceptor invokeInterceptor) {
- var node = new Node(
+ private Node newNode(NetworkAddress address, NodeFinder nodeFinder) {
+ return newNode(address, nodeFinder, null, defaultNodeAttributesConfiguration);
+ }
+
+ private Node newNode(
+ NetworkAddress address,
+ NodeFinder nodeFinder,
+ @Nullable Node.InvokeInterceptor invokeInterceptor,
+ NodeAttributesConfiguration nodeAttributesConfiguration
+ ) {
+ return new Node(
testInfo,
- NODE_ADDRESSES.get(idx),
- NODE_FINDER,
+ address,
+ nodeFinder,
workDir,
placementDriver,
systemConfiguration,
raftConfiguration,
- nodeAttributesConfigurations.get(idx),
+ nodeAttributesConfiguration,
storageConfiguration,
metaStorageConfiguration,
replicationConfiguration,
@@ -227,18 +245,6 @@ abstract class ItAbstractColocationTest extends IgniteAbstractTest {
invokeInterceptor,
gcConfiguration
);
-
- nodes.put(idx, node);
-
- return node;
- }
-
- Node getNode(int nodeIndex) {
- return nodes.get(nodeIndex);
- }
-
- Node getNode(String nodeName) {
- return nodes.values().stream().filter(n -> n.name.equals(nodeName)).findFirst().orElseThrow();
}
static void createZone(Node node, String zoneName, int partitions, int replicas) {
@@ -271,6 +277,14 @@ abstract class ItAbstractColocationTest extends IgniteAbstractTest {
);
}
+ Node getNode(int nodeIndex) {
+ return cluster.get(nodeIndex);
+ }
+
+ Node getNode(String nodeName) {
+ return cluster.stream().filter(n -> n.name.equals(nodeName)).findFirst().orElseThrow();
+ }
+
void setPrimaryReplica(Node node, @Nullable ZonePartitionId zonePartitionId) {
ClusterNode newPrimaryReplicaNode = node.clusterService.topologyService().localMember();
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItCatalogCompactionTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItCatalogCompactionTest.java
new file mode 100644
index 00000000000..43f6d7892d6
--- /dev/null
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItCatalogCompactionTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.partition.replicator;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
+import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.FINISHED;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.partition.replicator.fixtures.Node;
+import org.apache.ignite.internal.partitiondistribution.Assignment;
+import org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils;
+import org.apache.ignite.internal.partitiondistribution.TokenizedAssignments;
+import org.apache.ignite.internal.partitiondistribution.TokenizedAssignmentsImpl;
+import org.apache.ignite.internal.storage.pagememory.PersistentPageMemoryStorageEngine;
+import org.apache.ignite.internal.table.TableTestUtils;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.table.KeyValueView;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
+/**
+ * Tests catalog compaction for colocation track.
+ */
+@Timeout(60)
+public class ItCatalogCompactionTest extends ItAbstractColocationTest {
+ /**
+ * Tests catalog compaction.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-24558")
+ public void testCatalogCompaction() throws Exception {
+ // How often we update the low water mark.
+ long lowWatermarkUpdateInterval = 500;
+ updateLowWatermarkConfiguration(lowWatermarkUpdateInterval * 2, lowWatermarkUpdateInterval);
+
+ // Prepare a single node cluster.
+ startCluster(1);
+ Node node = getNode(0);
+
+ List<Set<Assignment>> assignments = PartitionDistributionUtils.calculateAssignments(
+ cluster.stream().map(n -> n.name).collect(toList()), 1, 1);
+
+ List<TokenizedAssignments> tokenizedAssignments = assignments.stream()
+ .map(a -> new TokenizedAssignmentsImpl(a, Integer.MIN_VALUE))
+ .collect(toList());
+
+ placementDriver.setPrimary(node.clusterService.topologyService().localMember());
+ placementDriver.setAssignments(tokenizedAssignments);
+
+ forceCheckpoint(node);
+
+ String zoneName = "test-zone";
+ createZone(node, zoneName, 1, 1);
+
+ int catalogVersion1 = getLatestCatalogVersion(node);
+
+ String tableName1 = "test_table_1";
+ createTable(node, zoneName, tableName1);
+
+ String tableName2 = "test_table_2";
+ createTable(node, zoneName, tableName2);
+
+ int tableId = TableTestUtils.getTableId(node.catalogManager, tableName2, node.hybridClock.nowLong());
+ TableViewInternal tableViewInternal = node.tableManager.table(tableId);
+ KeyValueView<Long, Integer> tableView = tableViewInternal.keyValueView(Long.class, Integer.class);
+
+ // Write 2 rows to the table.
+ Map<Long, Integer> valuesToPut = Map.of(0L, 0, 1L, 1);
+ assertDoesNotThrow(() -> tableView.putAll(null, valuesToPut));
+
+ forceCheckpoint(node);
+
+ int catalogVersion2 = getLatestCatalogVersion(node);
+ assertThat("The catalog version did not changed [initial=" + catalogVersion1 + ", latest=" + catalogVersion2 + "].",
+ catalogVersion2, greaterThan(catalogVersion1));
+
+ expectEarliestCatalogVersion(node, catalogVersion2 - 1);
+ }
+
+ private static void expectEarliestCatalogVersion(Node node, int expectedVersion) throws Exception {
+ boolean result = waitForCondition(() -> getEarliestCatalogVersion(node) == expectedVersion, 10_000);
+
+ assertTrue(result,
+ "Failed to wait for the expected catalog version [expected=" + expectedVersion
+ + ", earliest=" + getEarliestCatalogVersion(node)
+ + ", latest=" + getLatestCatalogVersion(node) + "].");
+ }
+
+ private static int getLatestCatalogVersion(Node node) {
+ Catalog catalog = getLatestCatalog(node);
+
+ return catalog.version();
+ }
+
+ private static int getEarliestCatalogVersion(Node node) {
+ CatalogManager catalogManager = node.catalogManager;
+
+ int ver = catalogManager.earliestCatalogVersion();
+
+ Catalog catalog = catalogManager.catalog(ver);
+
+ Objects.requireNonNull(catalog);
+
+ return catalog.version();
+ }
+
+ private static Catalog getLatestCatalog(Node node) {
+ CatalogManager catalogManager = node.catalogManager;
+
+ int ver = catalogManager.activeCatalogVersion(node.hybridClock.nowLong());
+
+ Catalog catalog = catalogManager.catalog(ver);
+
+ Objects.requireNonNull(catalog);
+
+ return catalog;
+ }
+
+ /**
+ * Start the new checkpoint immediately on the provided node.
+ *
+ * @param node Node to start the checkpoint on.
+ */
+ private void forceCheckpoint(Node node) {
+ PersistentPageMemoryStorageEngine storageEngine = (PersistentPageMemoryStorageEngine) node
+ .dataStorageManager()
+ .engineByStorageProfile(DEFAULT_STORAGE_PROFILE);
+
+ assertThat(storageEngine.checkpointManager().forceCheckpoint("test-reason").futureFor(FINISHED),
+ willSucceedIn(10, SECONDS));
+ }
+
+ /**
+ * Update low water mark configuration.
+ *
+ * @param dataAvailabilityTime Data availability time.
+ * @param updateInterval Update interval.
+ */
+ private void updateLowWatermarkConfiguration(long dataAvailabilityTime, long updateInterval) {
+ CompletableFuture<?> updateFuture = gcConfiguration.lowWatermark().change(change -> {
+ change.changeDataAvailabilityTime(dataAvailabilityTime);
+ change.changeUpdateInterval(updateInterval);
+ });
+
+ assertThat(updateFuture, willSucceedFast());
+ }
+}
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItKeyValueViewOperationsTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItKeyValueViewOperationsTest.java
index 5b436591545..d53b181b42f 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItKeyValueViewOperationsTest.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItKeyValueViewOperationsTest.java
@@ -39,7 +39,6 @@ import org.apache.ignite.lang.NullableValue;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.tx.Transaction;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
/**
* KeyValueView operations tests within colocation track.
@@ -49,11 +48,11 @@ public class ItKeyValueViewOperationsTest extends ItAbstractColocationTest {
private Node node;
@Test
- public void testKvOperations(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ public void testKvOperations() throws Exception {
+ startCluster(3);
Assignment replicaAssignment = (Assignment) calculateAssignmentForPartition(
- nodes.values().stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
+ cluster.stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
node = getNode(replicaAssignment.consistentId());
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
index 33298c1f4ab..278099ea6fd 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
@@ -19,22 +19,16 @@ package org.apache.ignite.internal.partition.replicator;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toList;
-import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.alterZone;
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.assertValueInStorage;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
-import static org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointState.FINISHED;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.flow.TestFlowUtils.subscribeToPublisher;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
import static org.apache.ignite.internal.util.ByteUtils.toByteArray;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.greaterThan;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -50,7 +44,6 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -62,19 +55,17 @@ import java.util.stream.IntStream;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.cluster.management.configuration.NodeAttributesConfiguration;
+import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.partition.replicator.fixtures.Node;
import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.partitiondistribution.Assignments;
-import org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils;
-import org.apache.ignite.internal.partitiondistribution.TokenizedAssignments;
-import org.apache.ignite.internal.partitiondistribution.TokenizedAssignmentsImpl;
import org.apache.ignite.internal.replicator.Replica;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.schema.BinaryRow;
-import org.apache.ignite.internal.storage.pagememory.PersistentPageMemoryStorageEngine;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableTestUtils;
import org.apache.ignite.internal.table.TableViewInternal;
@@ -87,7 +78,6 @@ import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.Table;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
/**
@@ -96,20 +86,28 @@ import org.junit.jupiter.api.Timeout;
@Timeout(60)
// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
+ @InjectConfiguration("mock.nodeAttributes: {region = US, storage = SSD}")
+ private static NodeAttributesConfiguration nodeAttributes1;
+
+ @InjectConfiguration("mock.nodeAttributes: {region = EU, storage = SSD}")
+ private static NodeAttributesConfiguration nodeAttributes2;
+
+ @InjectConfiguration("mock.nodeAttributes: {region = UK, storage = SSD}")
+ private static NodeAttributesConfiguration nodeAttributes3;
+
@Disabled("https://issues.apache.org/jira/browse/IGNITE-24374")
@Test
- public void testZoneReplicaListener(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ public void testZoneReplicaListener() throws Exception {
+ startCluster(3);
Assignment replicaAssignment = (Assignment) calculateAssignmentForPartition(
- nodes.values().stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
+ cluster.stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
Node node = getNode(replicaAssignment.consistentId());
placementDriver.setPrimary(node.clusterService.topologyService().localMember());
createZone(node, "test_zone", 1, 1);
- DistributionZonesTestUtil.getZoneId(node.catalogManager, "test_zone", node.hybridClock.nowLong());
long key = 1;
@@ -150,8 +148,8 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- void testAlterReplicaTrigger(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testAlterReplicaTrigger() throws Exception {
+ startCluster(3);
Node node = getNode(0);
@@ -168,7 +166,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
@@ -187,8 +185,8 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- void testAlterReplicaTriggerDefaultZone(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testAlterReplicaTriggerDefaultZone() throws Exception {
+ startCluster(3);
Node node = getNode(0);
@@ -224,8 +222,8 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- void testAlterReplicaExtensionTrigger(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testAlterReplicaExtensionTrigger() throws Exception {
+ startCluster(3);
Node node = getNode(0);
@@ -259,14 +257,15 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
}
@Test
- void testAlterFilterTrigger(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testAlterFilterTrigger() throws Exception {
+ List<NodeAttributesConfiguration> nodeAttributesConfigurations = List.of(nodeAttributes1, nodeAttributes2, nodeAttributes3);
+ startCluster(3, nodeAttributesConfigurations);
Node node = getNode(0);
@@ -285,7 +284,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
@@ -300,17 +299,17 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- Set.of(nodes.get(0).name),
+ Set.of(cluster.get(0).name),
20_000L
);
}
@Test
- void testTableReplicaListenersCreationAfterRebalance(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testTableReplicaListenersCreationAfterRebalance() throws Exception {
+ startCluster(3);
Assignment replicaAssignment = (Assignment) calculateAssignmentForPartition(
- nodes.values().stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
+ cluster.stream().map(n -> n.name).collect(toList()), 0, 1, 1).toArray()[0];
Node node = getNode(replicaAssignment.consistentId());
@@ -335,14 +334,14 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- void testTableReplicaListenersRemoveAfterRebalance(TestInfo testInfo) throws Exception {
+ void testTableReplicaListenersRemoveAfterRebalance() throws Exception {
String zoneName = "TEST_ZONE";
String tableName = "TEST_TABLE";
- startNodes(testInfo, 3);
+ startCluster(3);
Assignment replicaAssignment = (Assignment) calculateAssignmentForPartition(
- nodes.values().stream().map(n -> n.name).collect(toList()), 0, 1, 3).toArray()[0];
+ cluster.stream().map(n -> n.name).collect(toList()), 0, 1, 3).toArray()[0];
Node node = getNode(replicaAssignment.consistentId());
@@ -366,18 +365,18 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
30_000L
));
- nodes.values().forEach(n -> checkNoDestroyPartitionStoragesInvokes(n, tableName, 0));
+ cluster.forEach(n -> checkNoDestroyPartitionStoragesInvokes(n, tableName, 0));
alterZone(node.catalogManager, zoneName, 1);
- nodes.values().stream().filter(n -> !replicaAssignment.consistentId().equals(n.name)).forEach(
+ cluster.stream().filter(n -> !replicaAssignment.consistentId().equals(n.name)).forEach(
n -> checkDestroyPartitionStoragesInvokes(n, tableName, 0));
}
@Test
- void testReplicaIsStartedOnNodeStart(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 3);
+ void testReplicaIsStartedOnNodeStart() throws Exception {
+ startCluster(3);
Node node0 = getNode(0);
@@ -394,20 +393,20 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
stopNode(2);
- Node node2 = startNode(testInfo, 2);
+ Node node2 = addNodeToCluster(2);
assertTrue(waitForCondition(() -> node2.replicaManager.isReplicaStarted(partId), 10_000L));
}
@Test
- void testStableAreWrittenAfterRestart(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 1);
+ void testStableAreWrittenAfterRestart() throws Exception {
+ startCluster(1);
Node node0 = getNode(0);
@@ -443,7 +442,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stopNode(0);
- startNodes(testInfo, 1);
+ startCluster(1);
node0 = getNode(0);
@@ -454,7 +453,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
@@ -462,8 +461,8 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- void testStableAreWrittenAfterRestartAndConcurrentStableUpdate(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 1);
+ void testStableAreWrittenAfterRestartAndConcurrentStableUpdate() throws Exception {
+ startCluster(1);
Node node0 = getNode(0);
@@ -501,11 +500,11 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stopNode(0);
- startNodes(testInfo, 1, (condition, success, failure) -> {
+ startCluster(1, (condition, success, failure) -> {
if (skipMetaStorageInvoke(success, stablePartAssignmentsKey(partId).toString())) {
reached.set(true);
- Node node = nodes.get(0);
+ Node node = cluster.get(0);
int catalogVersion = node.catalogManager.latestCatalogVersion();
long timestamp = node.catalogManager.catalog(catalogVersion).time();
@@ -517,7 +516,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
return null;
- });
+ }, null);
node0 = getNode(0);
@@ -530,7 +529,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
stablePartAssignmentsKey(partId),
(v) -> Assignments.fromBytes(v).nodes()
.stream().map(Assignment::consistentId).collect(Collectors.toSet()),
- nodes.values().stream().map(n -> n.name).collect(Collectors.toSet()),
+ cluster.stream().map(n -> n.name).collect(Collectors.toSet()),
20_000L
);
@@ -538,18 +537,17 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- public void testTableEstimatedSize(TestInfo testInfo) throws Exception {
- startNodes(testInfo, 1);
+ public void testTableEstimatedSize() throws Exception {
+ startCluster(1);
Assignment replicaAssignment1 = (Assignment) calculateAssignmentForPartition(
- nodes.values().stream().map(n -> n.name).collect(toList()), 0, 2, 1).toArray()[0];
+ cluster.stream().map(n -> n.name).collect(toList()), 0, 2, 1).toArray()[0];
Node node = getNode(replicaAssignment1.consistentId());
placementDriver.setPrimary(node.clusterService.topologyService().localMember());
createZone(node, "test_zone", 2, 1);
- DistributionZonesTestUtil.getZoneId(node.catalogManager, "test_zone", node.hybridClock.nowLong());
{
createTable(node, "test_zone", "test_table_1");
@@ -596,16 +594,15 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
}
@Test
- public void testScanCloseReplicaRequest(TestInfo testInfo) throws Exception {
+ public void testScanCloseReplicaRequest() throws Exception {
// Prepare a single node cluster.
- startNodes(testInfo, 1);
+ startCluster(1);
Node node = getNode(0);
placementDriver.setPrimary(node.clusterService.topologyService().localMember());
// Prepare a zone.
String zoneName = "test_zone";
createZone(node, zoneName, 1, 1);
- DistributionZonesTestUtil.getZoneId(node.catalogManager, zoneName, node.hybridClock.nowLong());
// Create a table to work with.
String tableName = "test_table";
@@ -659,97 +656,6 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
assertDoesNotThrow(tx::commit);
}
- @Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-24558")
- public void testCatalogCompaction(TestInfo testInfo) throws Exception {
- // How often we update the low water mark.
- long lowWatermarkUpdateInterval = 500;
- updateLowWatermarkConfiguration(lowWatermarkUpdateInterval * 2, lowWatermarkUpdateInterval);
-
- // Prepare a single node cluster.
- startNodes(testInfo, 1);
- Node node = getNode(0);
-
- List<Set<Assignment>> assignments = PartitionDistributionUtils.calculateAssignments(
- nodes.values().stream().map(n -> n.name).collect(toList()), 1, 1);
-
- List<TokenizedAssignments> tokenizedAssignments = assignments.stream()
- .map(a -> new TokenizedAssignmentsImpl(a, Integer.MIN_VALUE))
- .collect(toList());
-
- placementDriver.setPrimary(node.clusterService.topologyService().localMember());
- placementDriver.setAssignments(tokenizedAssignments);
-
- forceCheckpoint(node);
-
- String zoneName = "test-zone";
- createZone(node, zoneName, 1, 1);
- DistributionZonesTestUtil.getZoneId(node.catalogManager, zoneName, node.hybridClock.nowLong());
-
- int catalogVersion1 = getLatestCatalogVersion(node);
-
- String tableName1 = "test_table_1";
- createTable(node, zoneName, tableName1);
-
- String tableName2 = "test_table_2";
- createTable(node, zoneName, tableName2);
-
- int tableId = TableTestUtils.getTableId(node.catalogManager, tableName2, node.hybridClock.nowLong());
- TableViewInternal tableViewInternal = node.tableManager.table(tableId);
- KeyValueView<Long, Integer> tableView = tableViewInternal.keyValueView(Long.class, Integer.class);
-
- // Write 2 rows to the table.
- Map<Long, Integer> valuesToPut = Map.of(0L, 0, 1L, 1);
- assertDoesNotThrow(() -> tableView.putAll(null, valuesToPut));
-
- forceCheckpoint(node);
-
- int catalogVersion2 = getLatestCatalogVersion(node);
- assertThat("The catalog version did not changed [initial=" + catalogVersion1 + ", latest=" + catalogVersion2 + ']',
- catalogVersion2, greaterThan(catalogVersion1));
-
- expectEarliestCatalogVersion(node, catalogVersion2 - 1);
- }
-
- private static void expectEarliestCatalogVersion(Node node, int expectedVersion) throws Exception {
- boolean result = waitForCondition(() -> getEarliestCatalogVersion(node) == expectedVersion, 10_000);
-
- assertTrue(result,
- "Failed to wait for the expected catalog version [expected=" + expectedVersion
- + ", earliest=" + getEarliestCatalogVersion(node)
- + ", latest=" + getLatestCatalogVersion(node) + ']');
- }
-
- private static int getLatestCatalogVersion(Node node) {
- Catalog catalog = getLatestCatalog(node);
-
- return catalog.version();
- }
-
- private static int getEarliestCatalogVersion(Node node) {
- CatalogManager catalogManager = node.catalogManager;
-
- int ver = catalogManager.earliestCatalogVersion();
-
- Catalog catalog = catalogManager.catalog(ver);
-
- Objects.requireNonNull(catalog);
-
- return catalog.version();
- }
-
- private static Catalog getLatestCatalog(Node node) {
- CatalogManager catalogManager = node.catalogManager;
-
- int ver = catalogManager.activeCatalogVersion(node.hybridClock.nowLong());
-
- Catalog catalog = catalogManager.catalog(ver);
-
- Objects.requireNonNull(catalog);
-
- return catalog;
- }
-
private static RemotelyTriggeredResource getVersionedStorageCursor(Node node, FullyQualifiedResourceId cursorId) {
return node.resourcesRegistry.resources().get(cursorId);
}
@@ -800,35 +706,6 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
.destroyTxStateStorage(partitionId);
}
- /**
- * Update low water mark configuration.
- *
- * @param dataAvailabilityTime Data availability time.
- * @param updateInterval Update interval.
- */
- private void updateLowWatermarkConfiguration(long dataAvailabilityTime, long updateInterval) {
- CompletableFuture<?> updateFuture = gcConfiguration.lowWatermark().change(change -> {
- change.changeDataAvailabilityTime(dataAvailabilityTime);
- change.changeUpdateInterval(updateInterval);
- });
-
- assertThat(updateFuture, willSucceedFast());
- }
-
- /**
- * Start the new checkpoint immediately on the provided node.
- *
- * @param node Node to start the checkpoint on.
- */
- private static void forceCheckpoint(Node node) {
- PersistentPageMemoryStorageEngine storageEngine = (PersistentPageMemoryStorageEngine) node
- .dataStorageManager()
- .engineByStorageProfile(DEFAULT_STORAGE_PROFILE);
-
- assertThat(storageEngine.checkpointManager().forceCheckpoint("test-reason").futureFor(FINISHED),
- willSucceedIn(10, SECONDS));
- }
-
@Test
public void enabledColocationTest() {
assertTrue(enabledColocation());
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItTransactionsVacuumTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItTransactionsVacuumTest.java
new file mode 100644
index 00000000000..40da5ae68aa
--- /dev/null
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItTransactionsVacuumTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.partition.replicator;
+
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteTransaction;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
+import static org.apache.ignite.internal.tx.impl.ResourceVacuumManager.RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil;
+import org.apache.ignite.internal.partition.replicator.fixtures.Node;
+import org.apache.ignite.internal.table.TableTestUtils;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.internal.testframework.SystemPropertiesExtension;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.tx.TransactionMeta;
+import org.apache.ignite.internal.tx.impl.ReadWriteTransactionImpl;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.tx.Transaction;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+// TODO: https://issues.apache.org/jira/browse/IGNITE-22522 remove this test after the switching to zone-based replication
+
+/**
+ * Tests transactions vacuum for colocation track.
+ */
+@ExtendWith(SystemPropertiesExtension.class)
+@WithSystemProperty(key = RESOURCE_VACUUM_INTERVAL_MILLISECONDS_PROPERTY, value = "500")
+@Timeout(60)
+public class ItTransactionsVacuumTest extends ItAbstractColocationTest {
+ /**
+ * Tests transactions vacuum.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTransactionsVacuum() throws Exception {
+ updateTxnResourceTtl(50L);
+
+ startCluster(1);
+ Node node = cluster.get(0);
+
+ String zoneName = "test-zone";
+ createZone(node, zoneName, 1, 1);
+ int zoneId = DistributionZonesTestUtil.getZoneId(node.catalogManager, zoneName, node.hybridClock.nowLong());
+
+ String tableName = "test_table";
+ createTable(node, zoneName, tableName);
+
+ int tableId = TableTestUtils.getTableId(node.catalogManager, tableName, node.hybridClock.nowLong());
+ TableViewInternal tableViewInternal = node.tableManager.table(tableId);
+ KeyValueView<Long, Integer> tableView = tableViewInternal.keyValueView(Long.class, Integer.class);
+
+ Transaction tx = node.transactions().begin();
+ UUID txId = txId(tx);
+
+ tableView.putAll(tx, Map.of(0L, 0, 1L, 1));
+
+ assertNotNull(volatileTxState(node, txId), "Volatile TX state is absent.");
+
+ assertNull(persistentTxState(node, zoneId, 0, txId), "Persistent TX state exists for non-completed TX.");
+
+ tx.commit();
+
+ // Check that volatile tx state is removed.
+ assertTrue(waitForCondition(() -> volatileTxState(node, txId) == null, 10_000));
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343 Enable this assertion.
+ // Check that persistent tx state is removed.
+ // assertTrue(waitForCondition(() -> persistentTxState(node, zoneId, 0, txId) == null, 10_000));
+ }
+
+ /**
+ * Updates txn resource time-to-live value..
+ *
+ * @param ttl Time-to-live value.
+ */
+ private void updateTxnResourceTtl(long ttl) {
+ CompletableFuture<?> updateFuture = txConfiguration.change(change -> change.changeTxnResourceTtl(ttl));
+
+ assertThat(updateFuture, willSucceedFast());
+ }
+
+ /**
+ * Returns transaction's meta from volatile storage.
+ *
+ * @param node Node.
+ * @param txId Transaction id.
+ * @return Transaction meta.
+ */
+ private static TransactionMeta volatileTxState(Node node, UUID txId) {
+ return node.txManager().stateMeta(txId);
+ }
+
+ /**
+ * Returns transaction's meta from persistent storage.
+ *
+ * @param node Node.
+ * @param zoneId Zone id.
+ * @param partId Partition id.
+ * @param txId Transaction id.
+ * @return Transaction meta.
+ */
+ private static TransactionMeta persistentTxState(Node node, int zoneId, int partId, UUID txId) {
+ return IgniteTestUtils.bypassingThreadAssertions(() -> node.txStatePartitionStorage(zoneId, partId).get(txId));
+ }
+
+ /**
+ * Returns transaction identifier for the given transaction.
+ *
+ * @param tx Transaction.
+ * @return Transaction id.
+ */
+ private static UUID txId(Transaction tx) {
+ return ((ReadWriteTransactionImpl) unwrapIgniteTransaction(tx)).id();
+ }
+}
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
index f4dbda25218..d9d86dca488 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
@@ -160,6 +160,7 @@ import org.apache.ignite.internal.tx.impl.HeapLockManager;
import org.apache.ignite.internal.tx.impl.IgniteTransactionsImpl;
import org.apache.ignite.internal.tx.impl.PublicApiThreadingIgniteTransactions;
import org.apache.ignite.internal.tx.impl.RemotelyTriggeredResourceRegistry;
+import org.apache.ignite.internal.tx.impl.ResourceVacuumManager;
import org.apache.ignite.internal.tx.impl.TransactionIdGenerator;
import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.internal.tx.impl.TxManagerImpl;
@@ -236,6 +237,9 @@ public class Node {
private final OutgoingSnapshotsManager outgoingSnapshotsManager;
+ /** Cleanup manager for tx resources. */
+ private final ResourceVacuumManager resourceVacuumManager;
+
/** The future have to be complete after the node start and all Meta storage watches are deployd. */
private CompletableFuture<Void> deployWatchesFut;
@@ -645,6 +649,16 @@ public class Node {
StorageUpdateConfiguration storageUpdateConfiguration = clusterConfigRegistry
.getConfiguration(StorageUpdateExtensionConfiguration.KEY).storageUpdate();
+ resourceVacuumManager = new ResourceVacuumManager(
+ name,
+ resourcesRegistry,
+ clusterService.topologyService(),
+ clusterService.messagingService(),
+ transactionInflights,
+ txManager,
+ lowWatermark
+ );
+
tableManager = new TableManager(
name,
registry,
@@ -761,7 +775,8 @@ public class Node {
outgoingSnapshotsManager,
partitionReplicaLifecycleManager,
tableManager,
- indexManager
+ indexManager,
+ resourceVacuumManager
)).thenComposeAsync(componentFuts -> {
CompletableFuture<Void> configurationNotificationFut = metaStorageManager.recoveryFinishedFuture()
.thenCompose(rev -> allOf(
@@ -842,4 +857,8 @@ public class Node {
public DataStorageManager dataStorageManager() {
return dataStorageMgr;
}
+
+ public TxManager txManager() {
+ return txManager;
+ }
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicationRaftCommandApplicator.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicationRaftCommandApplicator.java
index edcd15a2f41..ef988ca388b 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicationRaftCommandApplicator.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ReplicationRaftCommandApplicator.java
@@ -47,13 +47,14 @@ public class ReplicationRaftCommandApplicator {
* The original exception is set as cause.</li>
* </ul>
*
- * @param cmd Raft command.
+ * @param command Raft command.
* @return Raft future or raft decorated future with command that was processed.
*/
- public CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
+ // TODO https://issues.apache.org/jira/browse/IGNITE-22522 Perhaps, this method is not needed and can be removed in the future
+ public CompletableFuture<Object> applyCommandWithExceptionHandling(Command command) {
CompletableFuture<Object> resultFuture = new CompletableFuture<>();
- raftCommandRunner.run(cmd).whenComplete((res, ex) -> {
+ raftCommandRunner.run(command).whenComplete((res, ex) -> {
if (ex != null) {
resultFuture.completeExceptionally(ex);
} else {
@@ -71,4 +72,14 @@ public class ReplicationRaftCommandApplicator {
}
});
}
+
+ /**
+ * Executes the given {@code command} on the corresponding replication group.
+ *
+ * @param command The command to be executed.
+ * @return A future with the execution result.
+ */
+ public CompletableFuture<?> applyCommand(Command command) {
+ return raftCommandRunner.run(command);
+ }
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
index a0903359294..451ac916088 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.partition.replicator.handlers.MinimumActiveTxTimeReplicaRequestHandler;
import org.apache.ignite.internal.partition.replicator.handlers.TxFinishReplicaRequestHandler;
+import org.apache.ignite.internal.partition.replicator.handlers.VacuumTxStateReplicaRequestHandler;
import org.apache.ignite.internal.partition.replicator.network.replication.ReadOnlyReplicaRequest;
import org.apache.ignite.internal.partition.replicator.network.replication.UpdateMinimumActiveTxBeginTimeReplicaRequest;
import org.apache.ignite.internal.partition.replicator.schema.ValidationSchemasSource;
@@ -47,6 +48,7 @@ import org.apache.ignite.internal.replicator.message.TableAware;
import org.apache.ignite.internal.schema.SchemaSyncService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.message.TxFinishReplicaRequest;
+import org.apache.ignite.internal.tx.message.VacuumTxStateReplicaRequest;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.VisibleForTesting;
@@ -68,6 +70,7 @@ public class ZonePartitionReplicaListener implements ReplicaListener {
// Replica request handlers.
private final TxFinishReplicaRequestHandler txFinishReplicaRequestHandler;
private final MinimumActiveTxTimeReplicaRequestHandler minimumActiveTxTimeReplicaRequestHandler;
+ private final VacuumTxStateReplicaRequestHandler vacuumTxStateReplicaRequestHandler;
/**
* The constructor.
@@ -104,6 +107,8 @@ public class ZonePartitionReplicaListener implements ReplicaListener {
minimumActiveTxTimeReplicaRequestHandler = new MinimumActiveTxTimeReplicaRequestHandler(
clockService,
raftCommandApplicator);
+
+ vacuumTxStateReplicaRequestHandler = new VacuumTxStateReplicaRequestHandler(raftCommandApplicator);
}
@Override
@@ -198,7 +203,10 @@ public class ZonePartitionReplicaListener implements ReplicaListener {
) {
// TODO https://issues.apache.org/jira/browse/IGNITE-24526
// Need to move the necessary part of PartitionReplicaListener#processRequest request processing here
- if (request instanceof UpdateMinimumActiveTxBeginTimeReplicaRequest) {
+
+ if (request instanceof VacuumTxStateReplicaRequest) {
+ return vacuumTxStateReplicaRequestHandler.handle((VacuumTxStateReplicaRequest) request);
+ } else if (request instanceof UpdateMinimumActiveTxBeginTimeReplicaRequest) {
return minimumActiveTxTimeReplicaRequestHandler.handle((UpdateMinimumActiveTxBeginTimeReplicaRequest) request);
} else if (request instanceof ReplicaSafeTimeSyncRequest) {
LOG.debug("Non table request is not supported by the zone partition yet " + request);
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/MinimumActiveTxTimeReplicaRequestHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/MinimumActiveTxTimeReplicaRequestHandler.java
index 8a3d35067f1..a478a19563a 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/MinimumActiveTxTimeReplicaRequestHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/MinimumActiveTxTimeReplicaRequestHandler.java
@@ -66,6 +66,6 @@ public class MinimumActiveTxTimeReplicaRequestHandler {
// The timestamp must increase monotonically, otherwise it will have to be
// stored on disk so that reordering does not occur after the node is restarted.
- return commandApplicator.applyCmdWithExceptionHandling(cmd);
+ return commandApplicator.applyCommandWithExceptionHandling(cmd);
}
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
index 264de7b9927..f1e93425664 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/TxFinishReplicaRequestHandler.java
@@ -314,7 +314,7 @@ public class TxFinishReplicaRequestHandler {
finishTxCmdBldr.commitTimestamp(commitTimestamp);
}
- return raftCommandApplicator.applyCmdWithExceptionHandling(finishTxCmdBldr.build());
+ return raftCommandApplicator.applyCommandWithExceptionHandling(finishTxCmdBldr.build());
}
private static List<ReplicationGroupIdMessage> toPartitionIdMessage(Collection<ReplicationGroupId> partitionIds) {
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/VacuumTxStateReplicaRequestHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/VacuumTxStateReplicaRequestHandler.java
new file mode 100644
index 00000000000..1fd44869045
--- /dev/null
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/handlers/VacuumTxStateReplicaRequestHandler.java
@@ -0,0 +1,59 @@
+/*
+ * 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.partition.replicator.handlers;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.partition.replicator.ReplicationRaftCommandApplicator;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.VacuumTxStateReplicaRequest;
+import org.apache.ignite.internal.tx.message.VacuumTxStatesCommand;
+
+/**
+ * Handler for {@link VacuumTxStateReplicaRequest}.
+ */
+public class VacuumTxStateReplicaRequestHandler {
+ /** Factory for creating transaction command messages. */
+ private static final TxMessagesFactory TX_MESSAGES_FACTORY = new TxMessagesFactory();
+
+ /** Applicator that applies RAFT command that is created by this handler. */
+ private final ReplicationRaftCommandApplicator commandApplicator;
+
+ /**
+ * Creates a new instance of {@link VacuumTxStateReplicaRequestHandler}.
+ *
+ * @param commandApplicator Applicator that applies RAFT command that is created by this handler.
+ */
+ public VacuumTxStateReplicaRequestHandler(ReplicationRaftCommandApplicator commandApplicator) {
+ this.commandApplicator = commandApplicator;
+ }
+
+ /**
+ * Handles {@link VacuumTxStateReplicaRequest}.
+ *
+ * @param request Request to handle.
+ * @return Future that will be completed when the request is handled.
+ */
+ public CompletableFuture<?> handle(VacuumTxStateReplicaRequest request) {
+ VacuumTxStatesCommand cmd = TX_MESSAGES_FACTORY.vacuumTxStatesCommand()
+ .txIds(request.transactionIds())
+ .build();
+
+
+ return commandApplicator.applyCommand(cmd);
+ }
+}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListener.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListener.java
index 018b109b55b..18792c51648 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListener.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListener.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.partition.replicator.network.command.FinishTxC
import org.apache.ignite.internal.partition.replicator.network.command.TableAwareCommand;
import org.apache.ignite.internal.partition.replicator.network.command.UpdateMinimumActiveTxBeginTimeCommand;
import org.apache.ignite.internal.partition.replicator.raft.handlers.FinishTxCommandHandler;
+import org.apache.ignite.internal.partition.replicator.raft.handlers.VacuumTxStatesCommandHandler;
import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionKey;
import org.apache.ignite.internal.partition.replicator.raft.snapshot.ZonePartitionKey;
import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.PartitionSnapshots;
@@ -46,6 +47,7 @@ import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.command.SafeTimePropagatingCommand;
import org.apache.ignite.internal.replicator.message.PrimaryReplicaChangeCommand;
import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.message.VacuumTxStatesCommand;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.internal.util.SafeTimeValuesTracker;
@@ -78,10 +80,11 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
private final Object commitedConfigurationLock = new Object();
+ private final OnSnapshotSaveHandler onSnapshotSaveHandler;
+
// Raft command handlers.
private final FinishTxCommandHandler finishTxCommandHandler;
-
- private final OnSnapshotSaveHandler onSnapshotSaveHandler;
+ private final VacuumTxStatesCommandHandler vacuumTxStatesCommandHandler;
/** Constructor. */
public ZonePartitionRaftListener(
@@ -97,6 +100,8 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
this.partitionsSnapshots = partitionsSnapshots;
this.partitionKey = new ZonePartitionKey(zonePartitionId.zoneId(), zonePartitionId.partitionId());
+ onSnapshotSaveHandler = new OnSnapshotSaveHandler(txStatePartitionStorage, storageIndexTracker);
+
// RAFT command handlers initialization.
finishTxCommandHandler = new FinishTxCommandHandler(
txStatePartitionStorage,
@@ -105,7 +110,7 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
txManager
);
- onSnapshotSaveHandler = new OnSnapshotSaveHandler(txStatePartitionStorage, storageIndexTracker);
+ vacuumTxStatesCommandHandler = new VacuumTxStatesCommandHandler(txStatePartitionStorage);
}
@Override
@@ -163,23 +168,15 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
} else if (command instanceof PrimaryReplicaChangeCommand) {
// This is a hack for tests, this command is not issued in production because no zone-wide placement driver exists yet.
// FIXME: https://issues.apache.org/jira/browse/IGNITE-24374
- tableProcessors.values().forEach(listener -> listener.processCommand(command, commandIndex, commandTerm, safeTimestamp));
-
- result = new IgniteBiTuple<>(null, true);
+ result = processCrossTableProcessorsCommand(command, commandIndex, commandTerm, safeTimestamp);
} else if (command instanceof TableAwareCommand) {
TablePartitionId tablePartitionId = ((TableAwareCommand) command).tablePartitionId().asTablePartitionId();
result = processTableAwareCommand(tablePartitionId, command, commandIndex, commandTerm, safeTimestamp);
+ } else if (command instanceof VacuumTxStatesCommand) {
+ result = vacuumTxStatesCommandHandler.handle((VacuumTxStatesCommand) command, commandIndex, commandTerm);
} else if (command instanceof UpdateMinimumActiveTxBeginTimeCommand) {
- result = new IgniteBiTuple<>(null, false);
-
- tableProcessors.values().forEach(processor -> {
- IgniteBiTuple<Serializable, Boolean> r = processor.processCommand(command, commandIndex, commandTerm, safeTimestamp);
- // Need to adjust the safe time if any of the table processors successfully handled the command.
- if (Boolean.TRUE.equals(r.get2())) {
- result.set2(Boolean.TRUE);
- }
- });
+ result = processCrossTableProcessorsCommand(command, commandIndex, commandTerm, safeTimestamp);
} else {
LOG.info("Message type " + command.getClass() + " is not supported by the zone partition RAFT listener yet");
@@ -203,6 +200,44 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
clo.result(result.get1());
}
+ /**
+ * Redirects the command to all raft table processors to process.
+ *
+ * @param command Command to process.
+ * @param commandIndex Command index.
+ * @param commandTerm Command term.
+ * @param safeTimestamp Safe timestamp.
+ * @return Tuple with the result of the command processing and a flag indicating whether the command was applied.
+ */
+ private IgniteBiTuple<Serializable, Boolean> processCrossTableProcessorsCommand(
+ WriteCommand command,
+ long commandIndex,
+ long commandTerm,
+ @Nullable HybridTimestamp safeTimestamp
+ ) {
+ IgniteBiTuple<Serializable, Boolean> result = new IgniteBiTuple<>(null, false);
+
+ tableProcessors.values().forEach(processor -> {
+ IgniteBiTuple<Serializable, Boolean> r = processor.processCommand(command, commandIndex, commandTerm, safeTimestamp);
+ // Need to adjust the safe time if any of the table processors successfully handled the command.
+ if (Boolean.TRUE.equals(r.get2())) {
+ result.set2(Boolean.TRUE);
+ }
+ });
+
+ return result;
+ }
+
+ /**
+ * Redirects the command to a particular raft table processor for the given {@code tablePartitionId}.
+ *
+ * @param tablePartitionId Table partition identifier.
+ * @param command Command to process.
+ * @param commandIndex Command index.
+ * @param commandTerm Command term.
+ * @param safeTimestamp Safe timestamp.
+ * @return Tuple with the result of the command processing and a flag indicating whether the command was applied.
+ */
private IgniteBiTuple<Serializable, Boolean> processTableAwareCommand(
TablePartitionId tablePartitionId,
WriteCommand command,
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/VacuumTxStatesCommandHandler.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/VacuumTxStatesCommandHandler.java
new file mode 100644
index 00000000000..779a9f0da2a
--- /dev/null
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/handlers/VacuumTxStatesCommandHandler.java
@@ -0,0 +1,59 @@
+/*
+ * 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.partition.replicator.raft.handlers;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.lang.IgniteBiTuple;
+import org.apache.ignite.internal.tx.message.VacuumTxStatesCommand;
+import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
+
+/**
+ * RAFT command handler that process {@link VacuumTxStatesCommand} commands.
+ */
+public class VacuumTxStatesCommandHandler {
+ /** Storage of transaction metadata. */
+ private final TxStatePartitionStorage txStatePartitionStorage;
+
+ /**
+ * Creates a new instance of the command handler.
+ *
+ * @param txStatePartitionStorage Transactions state storage.
+ */
+ public VacuumTxStatesCommandHandler(TxStatePartitionStorage txStatePartitionStorage) {
+ this.txStatePartitionStorage = txStatePartitionStorage;
+ }
+
+ /**
+ * Handles {@link VacuumTxStatesCommand} command.
+ *
+ * @param command Command to be processed.
+ * @param commandIndex Command index.
+ * @param commandTerm Command term.
+ * @return Tuple with the result of the command processing and a flag indicating whether the command was applied.
+ */
+ public IgniteBiTuple<Serializable, Boolean> handle(VacuumTxStatesCommand command, long commandIndex, long commandTerm) {
+ // Skips the write command because the storage has already executed it.
+ if (commandIndex <= txStatePartitionStorage.lastAppliedIndex()) {
+ return new IgniteBiTuple<>(null, false);
+ }
+
+ txStatePartitionStorage.removeAll(command.txIds(), commandIndex, commandTerm);
+
+ return new IgniteBiTuple<>(null, true);
+ }
+}
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReplicaMessageUtils.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReplicaMessageUtils.java
index fe622eee09a..b07bfe6f86b 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReplicaMessageUtils.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReplicaMessageUtils.java
@@ -68,7 +68,10 @@ public class ReplicaMessageUtils {
ReplicaMessagesFactory messagesFactory,
ReplicationGroupId replicationGroupId
) {
- assert replicationGroupId instanceof TablePartitionId || replicationGroupId instanceof ZonePartitionId : replicationGroupId;
+ assert replicationGroupId instanceof TablePartitionId || replicationGroupId instanceof ZonePartitionId :
+ "Unexpected type of replication group identifier [class=" + replicationGroupId.getClass().getSimpleName()
+ + ", value=" + replicationGroupId
+ + ", requiredType=TablePartitionId or ZonePartitionId].";
if (replicationGroupId instanceof TablePartitionId) {
return toTablePartitionIdMessage(messagesFactory, (TablePartitionId) replicationGroupId);
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractColocationBenchmark.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractColocationBenchmark.java
index a54a51f81f2..41b6bfa8027 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractColocationBenchmark.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractColocationBenchmark.java
@@ -111,7 +111,7 @@ public class AbstractColocationBenchmark extends AbstractMultiNodeBenchmark {
public void nodeSetUp() throws Exception {
boolean colocationFeatureEnabled = enableColocationFeature();
- // Enable/disable collocation feature.
+ // Enable/disable colocation feature.
System.setProperty(COLOCATION_FEATURE_FLAG, Boolean.toString(colocationFeatureEnabled));
// Start the cluster and initialize it.
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java
index 1838c29bcec..27bf4635af8 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java
@@ -112,7 +112,7 @@ class PartitionStatesMetricSource implements MetricSource {
long[] count = {0};
// When receiving/loading metrics for all tables, the total complexity will be O(N*N) where N is the number of tables (for
- // simplicity). While this is done intentionally, in the future it will be necessary to optimize this or wait for the collocation
+ // simplicity). While this is done intentionally, in the future it will be necessary to optimize this or wait for the colocation
// of table partitions within one distribution zone.
disasterRecoveryManager.raftManager.forEach((raftNodeId, raftGroupService) -> {
if (raftNodeId.groupId() instanceof TablePartitionId) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index 499fb2558e9..c975c1798fe 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.table.distributed.raft;
import static java.util.Objects.requireNonNull;
import static org.apache.ignite.internal.hlc.HybridTimestamp.NULL_HYBRID_TIMESTAMP;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
import static org.apache.ignite.internal.table.distributed.TableUtils.indexIdsAtRwTxBeginTs;
import static org.apache.ignite.internal.table.distributed.TableUtils.indexIdsAtRwTxBeginTsOrNull;
import static org.apache.ignite.internal.table.distributed.index.MetaIndexStatus.BUILDING;
@@ -55,6 +56,7 @@ import org.apache.ignite.internal.partition.replicator.raft.OnSnapshotSaveHandle
import org.apache.ignite.internal.partition.replicator.raft.RaftTableProcessor;
import org.apache.ignite.internal.partition.replicator.raft.RaftTxFinishMarker;
import org.apache.ignite.internal.partition.replicator.raft.handlers.FinishTxCommandHandler;
+import org.apache.ignite.internal.partition.replicator.raft.handlers.VacuumTxStatesCommandHandler;
import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionDataStorage;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.RaftGroupConfiguration;
@@ -130,10 +132,9 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
// Raft command handlers.
private final RaftTxFinishMarker txFinisher;
-
private final FinishTxCommandHandler finishTxCommandHandler;
-
private final MinimumActiveTxTimeCommandHandler minimumActiveTxTimeCommandHandler;
+ private final VacuumTxStatesCommandHandler vacuumTxStatesCommandHandler;
/** Constructor. */
public PartitionListener(
@@ -165,6 +166,7 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
// RAFT command handlers initialization.
TablePartitionId tablePartitionId = new TablePartitionId(storage.tableId(), storage.partitionId());
txFinisher = new RaftTxFinishMarker(txManager);
+
finishTxCommandHandler = new FinishTxCommandHandler(
txStatePartitionStorage,
tablePartitionId,
@@ -174,6 +176,8 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
storage,
tablePartitionId,
minTimeCollectorService);
+
+ vacuumTxStatesCommandHandler = new VacuumTxStatesCommandHandler(txStatePartitionStorage);
}
@Override
@@ -246,7 +250,7 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
long commandTerm,
@Nullable HybridTimestamp safeTimestamp
) {
- IgniteBiTuple<Serializable, Boolean> result;
+ IgniteBiTuple<Serializable, Boolean> result = null;
if (command instanceof UpdateCommand) {
result = handleUpdateCommand((UpdateCommand) command, commandIndex, commandTerm, safeTimestamp);
@@ -263,10 +267,14 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
} else if (command instanceof PrimaryReplicaChangeCommand) {
result = handlePrimaryReplicaChangeCommand((PrimaryReplicaChangeCommand) command, commandIndex, commandTerm);
} else if (command instanceof VacuumTxStatesCommand) {
- result = handleVacuumTxStatesCommand((VacuumTxStatesCommand) command, commandIndex, commandTerm);
+ if (!enabledColocation()) {
+ result = vacuumTxStatesCommandHandler.handle((VacuumTxStatesCommand) command, commandIndex, commandTerm);
+ }
} else if (command instanceof UpdateMinimumActiveTxBeginTimeCommand) {
result = minimumActiveTxTimeCommandHandler.handle((UpdateMinimumActiveTxBeginTimeCommand) command, commandIndex);
- } else {
+ }
+
+ if (result == null) {
throw new AssertionError("Unknown command type [command=" + command.toStringForLightLogging() + ']');
}
@@ -630,28 +638,6 @@ public class PartitionListener implements RaftGroupListener, RaftTableProcessor
return new IgniteBiTuple<>(null, true);
}
- /**
- * Handler for {@link VacuumTxStatesCommand}.
- *
- * @param cmd Command.
- * @param commandIndex Command index.
- * @param commandTerm Command term.
- */
- private IgniteBiTuple<Serializable, Boolean> handleVacuumTxStatesCommand(
- VacuumTxStatesCommand cmd,
- long commandIndex,
- long commandTerm
- ) {
- // Skips the write command because the storage has already executed it.
- if (commandIndex <= storage.lastAppliedIndex()) {
- return new IgniteBiTuple<>(null, false);
- }
-
- txStatePartitionStorage.removeAll(cmd.txIds(), commandIndex, commandTerm);
-
- return new IgniteBiTuple<>(null, true);
- }
-
private static <T extends Comparable<T>> void updateTrackerIgnoringTrackerClosedException(
PendingComparableValuesTracker<T, Void> tracker,
T newValue
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/handlers/MinimumActiveTxTimeCommandHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/handlers/MinimumActiveTxTimeCommandHandler.java
index 14f0f5c3bcf..2c996456e50 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/handlers/MinimumActiveTxTimeCommandHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/handlers/MinimumActiveTxTimeCommandHandler.java
@@ -61,17 +61,17 @@ public class MinimumActiveTxTimeCommandHandler {
/**
* Handles {@link UpdateMinimumActiveTxBeginTimeCommand} command.
*
- * @param cmd Command to be processed.
+ * @param command Command to be processed.
* @param commandIndex Command index.
* @return Tuple with the result of the command processing and a flag indicating whether the command was applied.
*/
- public IgniteBiTuple<Serializable, Boolean> handle(UpdateMinimumActiveTxBeginTimeCommand cmd, long commandIndex) {
+ public IgniteBiTuple<Serializable, Boolean> handle(UpdateMinimumActiveTxBeginTimeCommand command, long commandIndex) {
// Skips the write command because the storage has already executed it.
if (commandIndex <= storage.lastAppliedIndex()) {
return new IgniteBiTuple<>(null, false);
}
- long timestamp = cmd.timestamp();
+ long timestamp = command.timestamp();
storage.flush(false)
.whenComplete((r, t) -> {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index de657a0c83d..0cd80842be8 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -111,6 +111,7 @@ import org.apache.ignite.internal.partition.replicator.ReplicaTxFinishMarker;
import org.apache.ignite.internal.partition.replicator.ReplicationRaftCommandApplicator;
import org.apache.ignite.internal.partition.replicator.handlers.MinimumActiveTxTimeReplicaRequestHandler;
import org.apache.ignite.internal.partition.replicator.handlers.TxFinishReplicaRequestHandler;
+import org.apache.ignite.internal.partition.replicator.handlers.VacuumTxStateReplicaRequestHandler;
import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory;
import org.apache.ignite.internal.partition.replicator.network.TimedBinaryRow;
import org.apache.ignite.internal.partition.replicator.network.command.BuildIndexCommand;
@@ -208,11 +209,9 @@ import org.apache.ignite.internal.tx.impl.FullyQualifiedResourceId;
import org.apache.ignite.internal.tx.impl.RemotelyTriggeredResourceRegistry;
import org.apache.ignite.internal.tx.message.TxCleanupRecoveryRequest;
import org.apache.ignite.internal.tx.message.TxFinishReplicaRequest;
-import org.apache.ignite.internal.tx.message.TxMessagesFactory;
import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
import org.apache.ignite.internal.tx.message.TxStateCommitPartitionRequest;
import org.apache.ignite.internal.tx.message.VacuumTxStateReplicaRequest;
-import org.apache.ignite.internal.tx.message.VacuumTxStatesCommand;
import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicaRequest;
import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicatedInfo;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
@@ -266,9 +265,6 @@ public class PartitionReplicaListener implements ReplicaListener {
/** Factory for creating replica command messages. */
private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
- /** Factory for creating transaction command messages. */
- private static final TxMessagesFactory TX_MESSAGES_FACTORY = new TxMessagesFactory();
-
/** Replication group id. */
private final TablePartitionId replicationGroupId;
@@ -359,6 +355,7 @@ public class PartitionReplicaListener implements ReplicaListener {
// Replica request handlers.
private final TxFinishReplicaRequestHandler txFinishReplicaRequestHandler;
private final MinimumActiveTxTimeReplicaRequestHandler minimumActiveTxTimeReplicaRequestHandler;
+ private final VacuumTxStateReplicaRequestHandler vacuumTxStateReplicaRequestHandler;
/**
* The constructor.
@@ -457,6 +454,8 @@ public class PartitionReplicaListener implements ReplicaListener {
clockService,
raftCommandApplicator);
+ vacuumTxStateReplicaRequestHandler = new VacuumTxStateReplicaRequestHandler(raftCommandApplicator);
+
prepareIndexBuilderTxRwOperationTracker();
}
@@ -883,7 +882,9 @@ public class PartitionReplicaListener implements ReplicaListener {
} else if (request instanceof TxStateCommitPartitionRequest) {
return processTxStateCommitPartitionRequest((TxStateCommitPartitionRequest) request);
} else if (request instanceof VacuumTxStateReplicaRequest) {
- return processVacuumTxStateReplicaRequest((VacuumTxStateReplicaRequest) request);
+ if (!enabledColocation()) {
+ return vacuumTxStateReplicaRequestHandler.handle((VacuumTxStateReplicaRequest) request);
+ }
} else if (request instanceof UpdateMinimumActiveTxBeginTimeReplicaRequest) {
if (!enabledColocation()) {
return minimumActiveTxTimeReplicaRequestHandler.handle((UpdateMinimumActiveTxBeginTimeReplicaRequest) request);
@@ -2499,7 +2500,7 @@ public class PartitionReplicaListener implements ReplicaListener {
}
private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
- return raftCommandApplicator.applyCmdWithExceptionHandling(cmd);
+ return raftCommandApplicator.applyCommandWithExceptionHandling(cmd);
}
/**
@@ -3969,14 +3970,6 @@ public class PartitionReplicaListener implements ReplicaListener {
return source == null ? null : new BinaryRowUpgrader(schemaRegistry, targetSchemaVersion).upgrade(source);
}
- private CompletableFuture<?> processVacuumTxStateReplicaRequest(VacuumTxStateReplicaRequest request) {
- VacuumTxStatesCommand cmd = TX_MESSAGES_FACTORY.vacuumTxStatesCommand()
- .txIds(request.transactionIds())
- .build();
-
- return raftCommandRunner.run(cmd);
- }
-
/**
* Operation unique identifier.
*/
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
index 5275852ed50..29cc1541f88 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PersistentTxStateVacuumizer.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.tx.impl;
import static java.util.stream.Collectors.toSet;
-import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toTablePartitionIdMessage;
+import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toReplicationGroupIdMessage;
import static org.apache.ignite.internal.util.CompletableFutures.allOf;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
@@ -38,7 +38,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.replicator.ReplicaService;
-import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.tx.message.TxMessagesFactory;
@@ -90,7 +90,11 @@ public class PersistentTxStateVacuumizer {
* @param txIds Transaction ids to vacuum; map of commit partition ids to sets of {@link VacuumizableTx}.
* @return A future, result is the set of successfully processed txn states and count of persistent states that were vacuumized.
*/
- public CompletableFuture<PersistentTxStateVacuumResult> vacuumPersistentTxStates(Map<TablePartitionId, Set<VacuumizableTx>> txIds) {
+ public CompletableFuture<PersistentTxStateVacuumResult> vacuumPersistentTxStates(
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343
+ // Should be changed to ZonePartitionId.
+ Map<ReplicationGroupId, Set<VacuumizableTx>> txIds
+ ) {
Set<UUID> successful = ConcurrentHashMap.newKeySet();
List<CompletableFuture<?>> futures = new ArrayList<>();
AtomicInteger vacuumizedPersistentTxnStatesCount = new AtomicInteger();
@@ -117,9 +121,16 @@ public class PersistentTxStateVacuumizer {
}
}
+ if (filteredTxIds.isEmpty()) {
+ // There is no need to send a request if there are no transaction metas to vacuum.
+ return nullCompletedFuture();
+ }
+
VacuumTxStateReplicaRequest request = TX_MESSAGES_FACTORY.vacuumTxStateReplicaRequest()
.enlistmentConsistencyToken(replicaMeta.getStartTime().longValue())
- .groupId(toTablePartitionIdMessage(REPLICA_MESSAGES_FACTORY, commitPartitionId))
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343
+ // Should be changed to ZonePartitionId.
+ .groupId(toReplicationGroupIdMessage(REPLICA_MESSAGES_FACTORY, commitPartitionId))
.transactionIds(filteredTxIds)
.build();
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
index c65d26ede8a..f6435a00514 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
@@ -982,7 +982,9 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
long vacuumObservationTimestamp = System.currentTimeMillis();
- return txStateVolatileStorage.vacuum(vacuumObservationTimestamp, txConfig.txnResourceTtl().value(),
+ return txStateVolatileStorage.vacuum(
+ vacuumObservationTimestamp,
+ txConfig.txnResourceTtl().value(),
persistentTxStateVacuumizer::vacuumPersistentTxStates);
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
index f5f2fb78818..32dcfb55fbb 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java
@@ -35,7 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.internal.tx.TxStateMeta;
@@ -144,7 +144,9 @@ public class VolatileTxStateMetaStorage {
public CompletableFuture<Void> vacuum(
long vacuumObservationTimestamp,
long txnResourceTtl,
- Function<Map<TablePartitionId, Set<VacuumizableTx>>, CompletableFuture<PersistentTxStateVacuumResult>> persistentVacuumOp
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343
+ // Should be changed to ZonePartitionId.
+ Function<Map<ReplicationGroupId, Set<VacuumizableTx>>, CompletableFuture<PersistentTxStateVacuumResult>> persistentVacuumOp
) {
LOG.info("Vacuum started [vacuumObservationTimestamp={}, txnResourceTtl={}].", vacuumObservationTimestamp, txnResourceTtl);
@@ -153,7 +155,9 @@ public class VolatileTxStateMetaStorage {
AtomicInteger alreadyMarkedTxnsCount = new AtomicInteger(0);
AtomicInteger skippedForFurtherProcessingUnfinishedTxnsCount = new AtomicInteger(0);
- Map<TablePartitionId, Set<VacuumizableTx>> txIds = new HashMap<>();
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343
+ // Should be changed to ZonePartitionId.
+ Map<ReplicationGroupId, Set<VacuumizableTx>> txIds = new HashMap<>();
Map<UUID, Long> cleanupCompletionTimestamps = new HashMap<>();
txStateMap.forEach((txId, meta) -> {
@@ -183,6 +187,8 @@ public class VolatileTxStateMetaStorage {
return null;
} else {
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24343
+ // Should be changed to ZonePartitionId.
Set<VacuumizableTx> ids = txIds.computeIfAbsent(meta0.commitPartitionId(), k -> new HashSet<>());
ids.add(new VacuumizableTx(txId, cleanupCompletionTimestamp));