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,13 @@ 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

@@ -135,8 +135,12 @@ private void assertRecoveryRequestForZoneTable(IgniteImpl node, String zoneName,
     }
 
     static void assertRecoveryRequestWasOnlyOne(IgniteImpl node) {
+        assertRecoveryRequestCount(node, 1);
+    }
+
+    static void assertRecoveryRequestCount(IgniteImpl node, int count) {
         assertEquals(
-                1,
+                count,
                 node
                         .metaStorageManager()
                         .getLocally(RECOVERY_TRIGGER_KEY.bytes(), 0L, Long.MAX_VALUE).size()
@@ -361,13 +365,13 @@ void stopNodes(Integer... nodes) {
         Arrays.stream(nodes).forEach(this::stopNode);
     }
 
-    static void changePartitionDistributionTimeout(IgniteImpl gatewayNode, int timeout) {
+    static void changePartitionDistributionTimeout(IgniteImpl gatewayNode, int timeoutSeconds) {
         CompletableFuture<Void> changeFuture = gatewayNode
                 .clusterConfiguration()
                 .getConfiguration(SystemDistributedExtensionConfiguration.KEY)
                 .system().change(c0 -> c0.changeProperties()
                         .createOrUpdate(PARTITION_DISTRIBUTION_RESET_TIMEOUT,
-                                c1 -> c1.changePropertyValue(String.valueOf(timeout)))
+                                c1 -> c1.changePropertyValue(String.valueOf(timeoutSeconds)))
                 );
 
         assertThat(changeFuture, willCompleteSuccessfully());
@@ -518,4 +522,16 @@ private static Row marshalKey(TableViewInternal table, Tuple key) {
     private static boolean isPrimaryReplicaHasChangedException(IgniteException cause) {
         return ExceptionUtils.extractCodeFrom(cause) == Replicator.REPLICA_MISS_ERR;
     }
+
+    void triggerManualReset(IgniteImpl node) {
+        CompletableFuture<?> updateFuture = node.disasterRecoveryManager().resetAllPartitions(
+                HA_ZONE_NAME,
+                SCHEMA_NAME,
+                HA_TABLE_NAME,
+                true,
+                -1
+        );
+
+        assertThat(updateFuture, willCompleteSuccessfully());
+    }
 }

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

@@ -17,13 +17,18 @@
 
 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;
@@ -37,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 CUSTOM_NODES_CONFIG = nodeConfig("{zone = custom}", null);
@@ -319,6 +326,126 @@ private void alterZoneSql(String filter, String zoneName) {
         executeSql(String.format("ALTER ZONE \"%s\" SET \"DATA_NODES_FILTER\" = '%s'", zoneName, filter));
     }
 
+    /**
+     * Test scenario.
+     * <ol>
+     *   <li>Create a zone in HA mode (node filter allows A, B, C)</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 5 minutes</li>
+     *   <li>Insert data and wait for replication to all nodes.</li>
+     *   <li>Stop a majority of nodes (B, C)</li>
+     *   <li>Change filter to allow (D, E, F) before 'partitionDistributionResetTimeout' expires</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 0 to trigger automatic reset</li>
+     *   <li>Verify the partition is on D,E,F</li>
+     *   <li>No data should be lost</li>
+     * </ol>
+     */
+    @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);
+
+        changePartitionDistributionTimeout(node, (int) TimeUnit.MINUTES.toSeconds(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);
+
+        changePartitionDistributionTimeout(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);
+    }
+
+    /**
+     * Test scenario.
+     * <ol>
+     *   <li>Create a zone in HA mode (profile filter allows A, B, C)</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 5 minutes</li>
+     *   <li>Insert data and wait for replication to all nodes.</li>
+     *   <li>Stop a majority of nodes (B, C)</li>
+     *   <li>Change filter to allow (D, E, F) before 'partitionDistributionResetTimeout' expires</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 0 to trigger automatic reset</li>
+     *   <li>Verify the partition is on D,E,F</li>
+     *   <li>No data should be lost</li>
+     * </ol>
+     */
+    @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);
+
+        changePartitionDistributionTimeout(node, (int) TimeUnit.MINUTES.toSeconds(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");
+
+        changePartitionDistributionTimeout(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

@@ -30,12 +30,14 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 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;
@@ -45,6 +47,7 @@
 import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
 import org.apache.ignite.internal.configuration.SystemDistributedExtensionConfiguration;
 import org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil;
+import org.apache.ignite.internal.metastorage.Entry;
 import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
 import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
 import org.apache.ignite.table.Table;
@@ -369,6 +372,73 @@ void testRestartNodesOneByOne() throws InterruptedException {
         assertValuesPresentOnNodes(node.clock().now(), node.tables().table(HA_TABLE_NAME), 0, 1, 2, 3, 4);
     }
 
+    /**
+     * Test scenario.
+     * <ol>
+     *   <li>Create a zone in HA mode (7 nodes, A, B, C, D, E, F, G)</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 5 minutes</li>
+     *   <li>Insert data and wait for replication to all nodes.</li>
+     *   <li>Stop a majority of nodes (4 nodes A, B, C, D)</li>
+     *   <li>Manually execute partition reset before 'partitionDistributionResetTimeout' expires</li>
+     *   <li>Wait for the partition to become available (E, F, G), no new writes</li>
+     *   <li>Set 'partitionDistributionResetTimeout' to 1 sec to trigger automatic reset</li>
+     *   <li>Verify the second (automatic) reset triggered</li>
+     *   <li>No data should be lost</li>
+     * </ol>
+     */
+    @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()));
+
+        changePartitionDistributionTimeout(node, (int) TimeUnit.MINUTES.toSeconds(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());
+
+        waitThatAllRebalancesHaveFinishedAndStableAssignmentsEqualsToExpected(node, HA_TABLE_NAME, PARTITION_IDS, threeNodes);
+
+        // This entry comes from the manual reset.
+        Entry manualResetTrigger = getRecoveryTriggerKey(node);
+        assertFalse(manualResetTrigger.empty());
+
+        long manualResetRev = manualResetTrigger.revision();
+
+        changePartitionDistributionTimeout(node, 1);
+
+        assertTrue(waitForCondition(() -> {
+                    Entry recoveryTrigger = getRecoveryTriggerKey(node);
+
+                    return !recoveryTrigger.empty() && recoveryTrigger.revision() > manualResetRev;
+                },
+                5_000
+        ));
+
+        assertRecoveryRequestForHaZoneTable(node);
+        assertRecoveryRequestCount(node, 2);
+
+        waitThatAllRebalancesHaveFinishedAndStableAssignmentsEqualsToExpected(node, HA_TABLE_NAME, PARTITION_IDS, threeNodes);
+    }
+
     @Test
     void testScaleDownTimerIsWorkingForHaZone() throws InterruptedException {
         IgniteImpl node = igniteImpl(0);