JAkutenshi commented on code in PR #5245:
URL: https://github.com/apache/ignite-3/pull/5245#discussion_r1962988900
##########
modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListener.java:
##########
@@ -163,23 +168,15 @@ private void
processWriteCommand(CommandClosure<WriteCommand> clo) {
} 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);
Review Comment:
I'm not sure that this block's code transformation equals to
`processCrossTableProcessorsCommand` in terms of output `IgniteBiTuple`. I
meant that the method could returns tuple with `false` second argument, but the
old one code couldn't. Is it ok?
##########
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.
Review Comment:
Just "the command"?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReplicaMessageUtils.java:
##########
@@ -68,7 +68,10 @@ public static ReplicationGroupIdMessage
toReplicationGroupIdMessage(
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]";
Review Comment:
Should we add a dot in the end of the log's message?
##########
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 AbstractColocationTest {
+ /**
+ * 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_1";
+ 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));
Review Comment:
It demands Roma's changes on `ZonePartitionId`?
##########
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 AbstractColocationTest {
+ /**
+ * 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_1";
Review Comment:
Is the postfix `_1` matters there?
##########
modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java:
##########
@@ -659,97 +656,6 @@ public void testScanCloseReplicaRequest(TestInfo testInfo)
throws Exception {
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());
Review Comment:
Just a comment: very weird that the statement didn't used but was presented.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/VolatileTxStateMetaStorage.java:
##########
@@ -144,7 +144,9 @@ public Collection<TxStateMeta> states() {
public CompletableFuture<Void> vacuum(
Review Comment:
Changes looks correct, but I can't figure out how do you found the place to
change `vaccuum` method? I mean, the requests handling aren't related to this,
aren't they?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -263,10 +267,14 @@ public IgniteBiTuple<Serializable, Boolean>
processCommand(
} 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) {
Review Comment:
We won't expect `null` result as any handler's result, the `null` value may
only occurs if there no a command support yet. With this changes a reader may
be confused "may there be an option for the result to be `null` from handler?".
Previous variant with `else` clause in my opinion do the job better, it throw
the exception only if we couldn't match a command without doubts about the
result's return value out of handlers.
##########
modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/AbstractColocationTest.java:
##########
@@ -68,107 +68,115 @@
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 {
+public class AbstractColocationTest extends IgniteAbstractTest {
Review Comment:
A minor comment: a class is named as `Abstract*` usually means it
`abstract`, but the class definition change is the exact opposite. May be
`Base*` prefix will be better?
##########
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 AbstractColocationTest {
+ /**
+ * 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) + ']');
Review Comment:
A dot in the end of log's message?
##########
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 AbstractColocationTest {
+ /**
+ * 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 + ']',
Review Comment:
..and an another dot?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
