IGNITE-24410 Implement filters and manual reset integration tests for HA

modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneHaManagerFilterTest.java

@@ -0,0 +1,31 @@
+/*
+ * 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.distributionzones;
+
+import org.apache.ignite.internal.catalog.descriptors.ConsistencyMode;
+
+/**
+ * Tests distribution zone manager interactions with data nodes filtering in an HA zone.
+ */
+public class DistributionZoneHaManagerFilterTest extends DistributionZoneManagerFilterTest {
+
+    @Override
+    protected ConsistencyMode consistencyMode() {
+        return ConsistencyMode.HIGH_AVAILABILITY;
+    }
+}

modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/DistributionZoneManagerFilterTest.java

@@ -25,13 +25,14 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.ignite.internal.catalog.descriptors.ConsistencyMode;
 import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
 import org.apache.ignite.internal.network.ClusterNodeImpl;
 import org.apache.ignite.network.NetworkAddress;
 import org.junit.jupiter.api.Test;
 
 /**
- * Tests distribution zone manager interactions with data nodes filtering.
+ * Tests distribution zone manager interactions with data nodes filtering in a SC zone.
  */
 public class DistributionZoneManagerFilterTest extends BaseDistributionZoneManagerTest {
     private static final LogicalNode A = new LogicalNode(
@@ -116,9 +117,14 @@ private void preparePrerequisites() throws Exception {
         topology.putNode(B);
         topology.putNode(C);
 
-        createZone(ZONE_NAME, IMMEDIATE_TIMER_VALUE, IMMEDIATE_TIMER_VALUE, filter, null, DEFAULT_STORAGE_PROFILE);
+        createZone(ZONE_NAME, IMMEDIATE_TIMER_VALUE, IMMEDIATE_TIMER_VALUE, filter, consistencyMode(), DEFAULT_STORAGE_PROFILE);
 
         assertDataNodesFromManager(distributionZoneManager, metaStorageManager::appliedRevision, catalogManager::latestCatalogVersion,
                 getZoneId(ZONE_NAME), Set.of(A, C), ZONE_MODIFICATION_AWAIT_TIMEOUT);
     }
+
+
+    protected ConsistencyMode consistencyMode() {
+        return ConsistencyMode.STRONG_CONSISTENCY;
+    }
 }

modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/AbstractHighAvailablePartitionsRecoveryTest.java

@@ -506,4 +506,28 @@ private static Row marshalKey(TableViewInternal table, Tuple key) {
     private static boolean isPrimaryReplicaHasChangedException(IgniteException cause) {
         return ExceptionUtils.extractCodeFrom(cause) == Replicator.REPLICA_MISS_ERR;
     }
+
+    void setDistributionResetTimeout(IgniteImpl node, long timeout) {
+        CompletableFuture<Void> changeFuture = node
+                .clusterConfiguration()
+                .getConfiguration(SystemDistributedExtensionConfiguration.KEY)
+                .system().change(c0 -> c0.changeProperties()
+                        .createOrUpdate(PARTITION_DISTRIBUTION_RESET_TIMEOUT,
+                                c1 -> c1.changePropertyValue(String.valueOf(timeout)))
+                );
+
+        assertThat(changeFuture, willCompleteSuccessfully());
+    }
+
+    void triggerManualReset(IgniteImpl node) {
+        CompletableFuture<?> updateFuture = node.disasterRecoveryManager().resetAllPartitions(
+                HA_ZONE_NAME,
+                SCHEMA_NAME,
+                HA_TABLE_NAME,
+                true,
+                0
+        );
+
+        assertThat(updateFuture, willCompleteSuccessfully());
+    }
 }

modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/ItHighAvailablePartitionsRecoveryByFilterUpdateTest.java

@@ -17,10 +17,22 @@
 
 package org.apache.ignite.internal.table.distributed.disaster;
 
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.is;
+
+import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.catalog.commands.AlterZoneCommand;
+import org.apache.ignite.internal.catalog.commands.AlterZoneCommandBuilder;
+import org.apache.ignite.internal.catalog.commands.StorageProfileParams;
+import org.apache.ignite.table.Table;
 import org.intellij.lang.annotations.Language;
 import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 /** Test suite for the cases with a recovery of the group replication factor after reset by zone filter update. */
@@ -30,6 +42,8 @@ public class ItHighAvailablePartitionsRecoveryByFilterUpdateTest extends Abstrac
 
     private static final String EU_ONLY_NODES_CONFIG = nodeConfig("{region = EU}", null);
 
+    private static final String US_ONLY_NODES_CONFIG = nodeConfig("{region = US}", null);
+
     private static final String GLOBAL_NODES_CONFIG = nodeConfig("{zone = global}", null);
 
     private static final String ROCKS_NODES_CONFIG = nodeConfig(null, "{lru_rocks.engine = rocksdb}");
@@ -126,6 +140,100 @@ private void alterZoneSql(String filter, String zoneName) {
         executeSql(String.format("ALTER ZONE \"%s\" SET \"DATA_NODES_FILTER\" = '%s'", zoneName, filter));
     }
 
+    @Disabled("https://issues.apache.org/jira/browse/IGNITE-24467")
+    @Test
+    void testResetAfterChangeFilters() throws InterruptedException {
+        startNode(1, EU_ONLY_NODES_CONFIG);
+        startNode(2, EU_ONLY_NODES_CONFIG);
+        startNode(3, US_ONLY_NODES_CONFIG);
+        startNode(4, US_ONLY_NODES_CONFIG);
+        startNode(5, US_ONLY_NODES_CONFIG);
+
+        String euFilter = "$[?(@.region == \"EU\")]";
+
+        Set<String> euNodes = nodeNames(0, 1, 2);
+
+        createHaZoneWithTable(euFilter, euNodes);
+
+        IgniteImpl node = igniteImpl(0);
+
+        setDistributionResetTimeout(node, TimeUnit.MINUTES.toMillis(5));
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, euNodes);
+
+        assertRecoveryKeyIsEmpty(node);
+
+        Table table = node.tables().table(HA_TABLE_NAME);
+
+        List<Throwable> errors = insertValues(table, 0);
+        assertThat(errors, is(empty()));
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 0, 1, 2);
+
+        stopNodes(1, 2);
+
+        String globalFilter = "$[?(@.region == \"US\")]";
+
+        alterZoneSql(globalFilter, HA_ZONE_NAME);
+
+        setDistributionResetTimeout(node, 0);
+
+        Set<String> usNodes = nodeNames(3, 4, 5);
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, usNodes);
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 3, 4, 5);
+    }
+
+    @Disabled("https://issues.apache.org/jira/browse/IGNITE-24467")
+    @Test
+    void testResetAfterChangeStorageProfiles() throws InterruptedException {
+        startNode(1, AIPERSIST_NODES_CONFIG);
+        startNode(2, AIPERSIST_NODES_CONFIG);
+        startNode(3, ROCKS_NODES_CONFIG);
+        startNode(4, ROCKS_NODES_CONFIG);
+        startNode(5, ROCKS_NODES_CONFIG);
+
+        Set<String> nodesWithAiProfile = nodeNames(0, 1, 2);
+
+        createHaZoneWithTableWithStorageProfile(2, "segmented_aipersist", nodesWithAiProfile);
+
+        IgniteImpl node = igniteImpl(0);
+
+        setDistributionResetTimeout(node, TimeUnit.MINUTES.toMillis(5));
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, nodesWithAiProfile);
+
+        assertRecoveryKeyIsEmpty(node);
+
+        Table table = node.tables().table(HA_TABLE_NAME);
+
+        List<Throwable> errors = insertValues(table, 0);
+        assertThat(errors, is(empty()));
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 0, 1, 2);
+
+        stopNodes(1, 2);
+
+        alterZoneStorageProfiles(node, HA_ZONE_NAME, "lru_rocks");
+
+        setDistributionResetTimeout(node, 0);
+
+        Set<String> usNodes = nodeNames(3, 4, 5);
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, usNodes);
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 3, 4, 5);
+    }
+
+    private void alterZoneStorageProfiles(IgniteImpl node, String zoneName, String storageProfile) {
+        AlterZoneCommandBuilder builder = AlterZoneCommand.builder().zoneName(zoneName);
+
+        builder.storageProfilesParams(List.of(StorageProfileParams.builder().storageProfile(storageProfile).build()));
+
+        assertThat(node.catalogManager().execute(builder.build()), willCompleteSuccessfully());
+    }
+
     private static String nodeConfig(
             @Nullable @Language("HOCON") String nodeAtrributes,
             @Nullable @Language("HOCON") String storageProfiles

modules/table/src/integrationTest/java/org/apache/ignite/internal/table/distributed/disaster/ItHighAvailablePartitionsRecoveryTest.java

@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.ignite.Ignite;
@@ -359,6 +360,44 @@ void testRestartNodesOneByOne() throws InterruptedException {
         assertValuesPresentOnNodes(node.clock().now(), node.tables().table(HA_TABLE_NAME), 0, 1, 2, 3, 4);
     }
 
+    @Test
+    void testManualRecovery() throws InterruptedException {
+        startNode(3);
+        startNode(4);
+        startNode(5);
+        startNode(6);
+
+        createHaZoneWithTable();
+
+        IgniteImpl node = igniteImpl(0);
+        Table table = node.tables().table(HA_TABLE_NAME);
+
+        List<Throwable> errors = insertValues(table, 0);
+        assertThat(errors, is(empty()));
+
+        setDistributionResetTimeout(node, TimeUnit.MINUTES.toMillis(5));
+
+        Set<String> allNodes = runningNodes().map(Ignite::name).collect(Collectors.toUnmodifiableSet());
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, allNodes);
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 0, 1, 2, 3, 4, 5, 6);
+
+        stopNodes(3, 4, 5, 6);
+
+        triggerManualReset(node);
+
+        Set<String> threeNodes = runningNodes().map(Ignite::name).collect(Collectors.toUnmodifiableSet());
+
+        waitAndAssertStableAssignmentsOfPartitionEqualTo(node, HA_TABLE_NAME, PARTITION_IDS, threeNodes);
+
+        setDistributionResetTimeout(node, TimeUnit.SECONDS.toMillis(0));
+
+        waitForCondition(() -> getRecoveryTriggerKey(node).empty(), 10_000);
+
+        assertValuesPresentOnNodes(node.clock().now(), table, 0, 1, 2);
+    }
+
     @Test
     void testScaleDownTimerIsWorkingForHaZone() throws InterruptedException {
         IgniteImpl node = igniteImpl(0);