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 47cfe883c23 IGNITE-25859 Implement pessimistic case of create zone on
unstable topology (#6430)
47cfe883c23 is described below
commit 47cfe883c230454abd9294607eaac1b9da52ae5d
Author: Mikhail Efremov <[email protected]>
AuthorDate: Mon Aug 25 15:16:32 2025 +0600
IGNITE-25859 Implement pessimistic case of create zone on unstable topology
(#6430)
---
modules/sql-engine/build.gradle | 1 +
.../internal/sql/api/ItSqlCreateZoneTest.java | 55 ++++++++++++++++++++++
.../ddl/ClusterWideStorageProfileValidator.java | 41 +++++++++++++---
.../DistributionZoneSqlToCommandConverterTest.java | 3 ++
4 files changed, 93 insertions(+), 7 deletions(-)
diff --git a/modules/sql-engine/build.gradle b/modules/sql-engine/build.gradle
index e7ea8ed53ee..0f9d3efad2a 100644
--- a/modules/sql-engine/build.gradle
+++ b/modules/sql-engine/build.gradle
@@ -124,6 +124,7 @@ dependencies {
integrationTestImplementation libs.netty.handler
integrationTestImplementation project(':ignite-api')
integrationTestImplementation project(':ignite-metastorage')
+ integrationTestImplementation project(':ignite-raft')
integrationTestImplementation project(':ignite-catalog')
integrationTestImplementation project(':ignite-transactions')
integrationTestImplementation project(':ignite-storage-api')
diff --git
a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlCreateZoneTest.java
b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlCreateZoneTest.java
index b3cf44ad20e..86151283230 100644
---
a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlCreateZoneTest.java
+++
b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlCreateZoneTest.java
@@ -21,13 +21,21 @@ import static
org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_AIMEM_
import static
org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_AIPERSIST_PROFILE_NAME;
import static
org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_ROCKSDB_PROFILE_NAME;
import static
org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_TEST_PROFILE_NAME;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static
org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCode;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.cluster.management.CmgGroupId;
+import org.apache.ignite.raft.jraft.rpc.RpcRequests.AppendEntriesRequest;
import org.apache.ignite.sql.SqlException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -71,6 +79,53 @@ class ItSqlCreateZoneTest extends
ClusterPerTestIntegrationTest {
assertDoesNotThrow(() -> createZoneQuery(0, EXTRA_PROFILE_NAME));
}
+ @Test
+ void
testCreateZoneSucceedWithCorrectStorageProfileOnDifferentNodeWithDistributedLogicalTopologyUpdate()
throws InterruptedException {
+ // Node 0 is CMG leader and Node 1 is a laggy query executor.
+ IgniteImpl node0 = unwrapIgniteImpl(node(0));
+ IgniteImpl node1 = unwrapIgniteImpl(cluster.startNode(1));
+
+ assertTrue(waitForCondition(
+ () ->
node1.logicalTopologyService().localLogicalTopology().nodes().size() == 2,
+ 10_000
+ ));
+
+ // Assert that we can't create the zone without a node with extra
profile.
+ assertThrowsWithCause(
+ () -> createZoneQuery(1, EXTRA_PROFILE_NAME),
+ SqlException.class,
+ "Some storage profiles don't exist [missedProfileNames=[" +
EXTRA_PROFILE_NAME + "]]."
+ );
+
+ // Node 1 won't see node 2 joined with extra profile because node 0 is
CMG leader and all CMG-related RAFT-replicated messages to
+ // node 1 will be dropped after the code below.
+ node0.dropMessages((recipient, msg) -> msg instanceof
AppendEntriesRequest
+ && ((AppendEntriesRequest)
msg).groupId().equals(CmgGroupId.INSTANCE.toString())
+ && node1.name().equals(recipient));
+
+ // Then start node 2 with the desired extra profile.
+ cluster.startNode(2, NODE_BOOTSTRAP_CFG_TEMPLATE_WITH_EXTRA_PROFILE);
+
+ // Check that Node 0 and 2 will see all three nodes in local logical
topologies.
+ assertTrue(waitForCondition(
+ () ->
unwrapIgniteImpl(node(0)).logicalTopologyService().localLogicalTopology().nodes().size()
== 3,
+ 10_000
+ ));
+
+ assertTrue(waitForCondition(
+ () ->
unwrapIgniteImpl(node(2)).logicalTopologyService().localLogicalTopology().nodes().size()
== 3,
+ 10_000
+ ));
+
+ // And we expect that node 1 won't see node 2 in its local logical
topology.
+ assertEquals(2,
node1.logicalTopologyService().localLogicalTopology().nodes().size());
+
+ // But still we're able to create zone with extra profile on node 2
because node 1 will try to ask CMG leader (node 0) directly over
+ // the network for its up-to-date leader's local logical topology and
check this snapshot's storage profiles that should
+ // extra profile because 2nd node was accepted to cluster by node 0
because it's the only CMG group voting member.
+ assertDoesNotThrow(() -> createZoneQuery(1, EXTRA_PROFILE_NAME));
+ }
+
@Test
void testCreateZoneFailedWithoutCorrectStorageProfileInCluster() {
assertThrowsWithCode(
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/ddl/ClusterWideStorageProfileValidator.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/ddl/ClusterWideStorageProfileValidator.java
index 367dd230ad7..221615a8801 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/ddl/ClusterWideStorageProfileValidator.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/ddl/ClusterWideStorageProfileValidator.java
@@ -17,8 +17,10 @@
package org.apache.ignite.internal.sql.engine.prepare.ddl;
-import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static
org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
import java.util.Collection;
@@ -49,14 +51,39 @@ public class ClusterWideStorageProfileValidator implements
StorageProfileValidat
localLogicalTopologySnapshot
);
- if (!missedStorageProfileNames.isEmpty()) {
- throw new SqlException(STMT_VALIDATION_ERR, format(
- "Some storage profiles don't exist
[missedProfileNames={}].",
- missedStorageProfileNames
- ));
+ if (missedStorageProfileNames.isEmpty()) {
+ return nullCompletedFuture();
}
- return completedFuture(null);
+ return logicalTopologyService.logicalTopologyOnLeader()
+ .thenApply(topologySnapshot ->
findStorageProfileNotPresentedInLogicalTopologySnapshot(
+ storageProfiles,
+ topologySnapshot
+ )).handle((missedProfileNames, e) -> {
+ if (e != null) {
+ String msg = format(
+ "Storage profiles {} don't exist in local
topology snapshot with profiles [{}], "
+ + "and distributed refresh
failed.",
+ missedStorageProfileNames,
+ localLogicalTopologySnapshot
+ .nodes()
+ .stream()
+ .map(LogicalNode::storageProfiles)
+ .collect(toSet())
+ );
+
+ throw new SqlException(INTERNAL_ERR, msg, e);
+ }
+
+ if (!missedProfileNames.isEmpty()) {
+ throw new SqlException(STMT_VALIDATION_ERR, format(
+ "Some storage profiles don't exist
[missedProfileNames={}].",
+ missedProfileNames
+ ));
+ }
+
+ return null;
+ });
}
private static Set<String>
findStorageProfileNotPresentedInLogicalTopologySnapshot(
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/DistributionZoneSqlToCommandConverterTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/DistributionZoneSqlToCommandConverterTest.java
index 96896c4be52..f88b632bb1c 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/DistributionZoneSqlToCommandConverterTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/ddl/DistributionZoneSqlToCommandConverterTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.sql.engine.prepare.ddl;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static
org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
import static
org.apache.ignite.internal.catalog.commands.CatalogUtils.INFINITE_TIMER_VALUE;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
@@ -113,6 +114,8 @@ public class DistributionZoneSqlToCommandConverterTest
extends AbstractDdlSqlToC
when(logicalTopologyService.localLogicalTopology()).thenReturn(defaultLogicalTopologySnapshot);
+
when(logicalTopologyService.logicalTopologyOnLeader()).thenReturn(completedFuture(defaultLogicalTopologySnapshot));
+
converter = new DdlSqlToCommandConverter(new
ClusterWideStorageProfileValidator(logicalTopologyService));
assertThat(ZoneOptionEnum.values().length, is(NUMERIC_OPTIONS.size() +
STRING_OPTIONS.size()));
