You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sanpwc (via GitHub)" <gi...@apache.org> on 2023/01/31 21:37:11 UTC
[GitHub] [ignite-3] sanpwc commented on a diff in pull request #1572: IGNITE-18088 Trigger rebalance on zone.dataNodes change.
sanpwc commented on code in PR #1572:
URL: https://github.com/apache/ignite-3/pull/1572#discussion_r1092413754
##########
modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityUtils.java:
##########
@@ -38,8 +37,8 @@ public class AffinityUtils {
* @param replicas Replicas count.
Review Comment:
javadoc lacks baselineNodes param. And it's no longer baselineNodes. I'd rather use dataNodes one.
##########
modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityUtils.java:
##########
@@ -48,32 +47,32 @@ public static List<Set<Assignment>> calculateAssignments(Collection<ClusterNode>
HashSet::new
);
- return affinityNodes.stream().map(AffinityUtils::clusterNodesToAssignments).collect(toList());
+ return affinityNodes.stream().map(AffinityUtils::consistentIdsToAssignments).collect(toList());
}
/**
* Calculates affinity assignments for a single partition.
*
- * @param baselineNodes Nodes.
+ * @param consistentIds Consistent ids of nodes.
* @param partition Partition id.
* @param replicas Replicas count.
- * @return List of assignments.
+ * @return Set of assignments.
*/
- public static Set<Assignment> calculateAssignmentForPartition(Collection<ClusterNode> baselineNodes, int partition, int replicas) {
- Set<ClusterNode> affinityNodes = RendezvousAffinityFunction.assignPartition(
+ public static Set<Assignment> calculateAssignmentForPartition(Collection<String> consistentIds, int partition, int replicas) {
Review Comment:
From my point of view consistentIds is a bit confusing, I'd rather use dataNodes with more meaningful explanation in @param section that will explicitly denote that it's **consistent** between restarts set of data nodes identifiers.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
Review Comment:
It's actually an integration rebalance test and not a distributionZones one, is that correct?
##########
modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZonesUtil.java:
##########
@@ -61,10 +62,27 @@ class DistributionZonesUtil {
new ByteArray(DISTRIBUTION_ZONES_LOGICAL_TOPOLOGY_VERSION);
/** ByteArray representation of {@link DistributionZonesUtil#DISTRIBUTION_ZONE_DATA_NODES_PREFIX}. */
- static ByteArray zoneDataNodesKey(int zoneId) {
+ public static ByteArray zoneDataNodesKey(int zoneId) {
return new ByteArray(DISTRIBUTION_ZONE_DATA_NODES_PREFIX + zoneId);
}
+ /** ByteArray representation of {@link DistributionZonesUtil#DISTRIBUTION_ZONE_DATA_NODES_PREFIX}. */
Review Comment:
Is it really valid to have such one-liners for methods?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
+ private static final IgniteLogger LOG = Loggers.forClass(ItDistributionZonesTest.class);
+
+ /**
+ * Nodes bootstrap configuration pattern.
+ *
+ * <p>rpcInstallSnapshotTimeout is changed to 10 seconds so that sporadic snapshot installation failures still
+ * allow tests pass thanks to retries.
+ */
+ private static final String NODE_BOOTSTRAP_CFG = "{\n"
+ + " network: {\n"
+ + " port:{},\n"
+ + " nodeFinder:{\n"
+ + " netClusterNodes: [ {} ]\n"
+ + " }\n"
+ + " },\n"
+ + " raft.rpcInstallSnapshotTimeout: 10000"
+ + "}";
+
+ @WorkDirectory
+ private Path workDir;
+
+ private Cluster cluster;
+
+ @BeforeEach
+ void createCluster(TestInfo testInfo) {
+ cluster = new Cluster(testInfo, workDir, NODE_BOOTSTRAP_CFG);
+ }
+
+ @AfterEach
+ @Timeout(60)
+ void shutdownCluster() {
+ cluster.shutdown();
+ }
+
+ @Test
+ @Disabled
+ void assingmentsChangingOnNodeLeaveNodeJoin() throws Exception {
+ cluster.startAndInit(4);
+
+ createTestTable();
Review Comment:
Could you please add a comment that we are creating the table with one partition and three replicas? Minor, though.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
+ private static final IgniteLogger LOG = Loggers.forClass(ItDistributionZonesTest.class);
+
+ /**
+ * Nodes bootstrap configuration pattern.
+ *
+ * <p>rpcInstallSnapshotTimeout is changed to 10 seconds so that sporadic snapshot installation failures still
+ * allow tests pass thanks to retries.
+ */
+ private static final String NODE_BOOTSTRAP_CFG = "{\n"
+ + " network: {\n"
+ + " port:{},\n"
+ + " nodeFinder:{\n"
+ + " netClusterNodes: [ {} ]\n"
+ + " }\n"
+ + " },\n"
+ + " raft.rpcInstallSnapshotTimeout: 10000"
+ + "}";
+
+ @WorkDirectory
+ private Path workDir;
+
+ private Cluster cluster;
+
+ @BeforeEach
+ void createCluster(TestInfo testInfo) {
+ cluster = new Cluster(testInfo, workDir, NODE_BOOTSTRAP_CFG);
+ }
+
+ @AfterEach
+ @Timeout(60)
Review Comment:
And what about timeout values, both 60 and 90. Is it also just a copy paste from ItTableRaftSnapshotsTest or we really expect given test to run up to 90 seconds?
##########
modules/affinity/src/test/java/org/apache/ignite/internal/affinity/RendezvousAffinityFunctionTest.java:
##########
@@ -104,29 +102,26 @@ public void testPartitionDistribution() {
}
@NotNull
- private List<ClusterNode> prepareNetworkTopology(int nodes) {
- var addr = new NetworkAddress("127.0.0.1", 121212);
-
+ private List<String> prepareNetworkTopology(int nodes) {
return IntStream.range(0, nodes)
.mapToObj(i -> "Node " + i)
- .map(name -> new ClusterNode(UUID.randomUUID().toString(), name, addr))
.collect(Collectors.toUnmodifiableList());
}
@Test
public void serializeAssignment() {
- int nodes = 50;
+ int nodeCount = 50;
int parts = 10_000;
int replicas = 4;
- List<ClusterNode> clusterNodes = prepareNetworkTopology(nodes);
+ List<String> nodes = prepareNetworkTopology(nodeCount);
- assertTrue(parts > nodes, "Partitions should be more that nodes");
+ assertTrue(parts > nodeCount, "Partitions should be more that nodes");
Review Comment:
Same as above mentioned by senior @alievmirza - than.
##########
modules/affinity/src/main/java/org/apache/ignite/internal/affinity/AffinityUtils.java:
##########
@@ -48,32 +47,32 @@ public static List<Set<Assignment>> calculateAssignments(Collection<ClusterNode>
HashSet::new
);
- return affinityNodes.stream().map(AffinityUtils::clusterNodesToAssignments).collect(toList());
+ return affinityNodes.stream().map(AffinityUtils::consistentIdsToAssignments).collect(toList());
}
/**
* Calculates affinity assignments for a single partition.
*
- * @param baselineNodes Nodes.
+ * @param consistentIds Consistent ids of nodes.
* @param partition Partition id.
* @param replicas Replicas count.
- * @return List of assignments.
+ * @return Set of assignments.
*/
- public static Set<Assignment> calculateAssignmentForPartition(Collection<ClusterNode> baselineNodes, int partition, int replicas) {
- Set<ClusterNode> affinityNodes = RendezvousAffinityFunction.assignPartition(
+ public static Set<Assignment> calculateAssignmentForPartition(Collection<String> consistentIds, int partition, int replicas) {
+ Set<String> affinityNodes = RendezvousAffinityFunction.assignPartition(
partition,
- new ArrayList<>(baselineNodes),
+ new ArrayList<>(consistentIds),
replicas,
null,
false,
null,
HashSet::new
);
- return clusterNodesToAssignments(affinityNodes);
+ return consistentIdsToAssignments(affinityNodes);
}
- private static Set<Assignment> clusterNodesToAssignments(Collection<ClusterNode> nodes) {
- return nodes.stream().map(node -> Assignment.forPeer(node.name())).collect(toSet());
+ private static Set<Assignment> consistentIdsToAssignments(Collection<String> nodes) {
+ return nodes.stream().map(node -> Assignment.forPeer(node)).collect(toSet());
Review Comment:
Let's use method reference here ;)
##########
modules/runner/build.gradle:
##########
@@ -114,6 +114,7 @@ dependencies {
integrationTestImplementation project(':ignite-metastorage')
integrationTestImplementation project(':ignite-table')
integrationTestImplementation project(':ignite-transactions')
+ integrationTestImplementation project(':ignite-distribution-zones')
Review Comment:
Do we really use distribution zones module within runner tests?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
+ private static final IgniteLogger LOG = Loggers.forClass(ItDistributionZonesTest.class);
+
+ /**
+ * Nodes bootstrap configuration pattern.
+ *
+ * <p>rpcInstallSnapshotTimeout is changed to 10 seconds so that sporadic snapshot installation failures still
Review Comment:
It looks like a copy paste from ItTableRaftSnapshotsTest. Do we really need 10 seconds adjust here?
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
+ private static final IgniteLogger LOG = Loggers.forClass(ItDistributionZonesTest.class);
+
+ /**
+ * Nodes bootstrap configuration pattern.
+ *
+ * <p>rpcInstallSnapshotTimeout is changed to 10 seconds so that sporadic snapshot installation failures still
+ * allow tests pass thanks to retries.
+ */
+ private static final String NODE_BOOTSTRAP_CFG = "{\n"
+ + " network: {\n"
+ + " port:{},\n"
+ + " nodeFinder:{\n"
+ + " netClusterNodes: [ {} ]\n"
+ + " }\n"
+ + " },\n"
+ + " raft.rpcInstallSnapshotTimeout: 10000"
+ + "}";
+
+ @WorkDirectory
+ private Path workDir;
+
+ private Cluster cluster;
+
+ @BeforeEach
+ void createCluster(TestInfo testInfo) {
+ cluster = new Cluster(testInfo, workDir, NODE_BOOTSTRAP_CFG);
+ }
+
+ @AfterEach
+ @Timeout(60)
+ void shutdownCluster() {
+ cluster.shutdown();
+ }
+
+ @Test
+ @Disabled
+ void assingmentsChangingOnNodeLeaveNodeJoin() throws Exception {
+ cluster.startAndInit(4);
+
+ createTestTable();
+
+ assertTrue(waitAssingments(List.of(
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2)
+ )));
+
+ TableImpl table = (TableImpl) cluster.node(0).tables().table("TEST");
+
+ BinaryRowEx key = new TupleMarshallerImpl(table.schemaView()).marshal(Tuple.create().set("id", 1));
+
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(0).node()).get());
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(1).node()).get());
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(2).node()).get());
+
+ putData();
+
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(0).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(1).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(2).node()).get());
+
+ try {
+ table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(3).node()).get();
+
+ fail();
+ } catch (Exception e) {
+ assertInstanceOf(ExecutionException.class, e);
+
+ assertInstanceOf(ReplicaUnavailableException.class, e.getCause());
+ }
+
+ cluster.knockOutNode(2, PARTITION_NETWORK);
+
+ assertTrue(waitAssingments(List.of(
+ Set.of(0, 1, 3),
+ Set.of(0, 1, 3),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 3)
+ )));
+
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(0).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(1).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(3).node()).get());
+
+ cluster.reanimateNode(2, PARTITION_NETWORK);
+
+ assertTrue(waitAssingments(List.of(
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2)
+ )));
+
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(0).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(1).node()).get());
+ assertNotNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(2).node()).get());
+
+ try {
+ table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(3).node()).get();
+
+ fail();
+ } catch (Exception e) {
+ assertInstanceOf(ExecutionException.class, e);
+
+ assertInstanceOf(ReplicaUnavailableException.class, e.getCause());
+ }
+ }
+
+ private boolean waitAssingments(List<Set<Integer>> nodes) throws InterruptedException {
+ return waitForCondition(() -> {
+ for (int i = 0; i < nodes.size(); i++) {
+ Set<Integer> excpectedAssignments = nodes.get(i);
+
+ ExtendedTableConfiguration table =
+ (ExtendedTableConfiguration) cluster.node(i)
+ .clusterConfiguration().getConfiguration(TablesConfiguration.KEY).tables().get("TEST");
+
+ byte[] assignmentsBytes = table.assignments().value();
+
+ Set<String> assignments;
+
+ if (assignmentsBytes != null) {
+ assignments = ((List<Set<Assignment>>) ByteUtils.fromBytes(assignmentsBytes)).get(0)
+ .stream().map(assignment -> assignment.consistentId()).collect(Collectors.toSet());
+ } else {
+ assignments = Collections.emptySet();
+ }
+
+ LOG.info("Assignments for node " + i + ": " + assignments);
+
+ if (!(excpectedAssignments.size() == assignments.size())
+ || !excpectedAssignments.stream().allMatch(node -> assignments.contains(cluster.node(node).name()))) {
Review Comment:
It's not a good idea to check whether node name contains the number and consider that as an equality relation. I'd use full node names for an expected value.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItDistributionZonesTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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 static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.Cluster.NodeKnockout.PARTITION_NETWORK;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+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 static org.junit.jupiter.api.Assertions.fail;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.Cluster;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.replicator.exception.ReplicaUnavailableException;
+import org.apache.ignite.internal.schema.BinaryRowEx;
+import org.apache.ignite.internal.schema.configuration.ExtendedTableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Test suite for rebalance process.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@Timeout(90)
+public class ItDistributionZonesTest {
+ private static final IgniteLogger LOG = Loggers.forClass(ItDistributionZonesTest.class);
+
+ /**
+ * Nodes bootstrap configuration pattern.
+ *
+ * <p>rpcInstallSnapshotTimeout is changed to 10 seconds so that sporadic snapshot installation failures still
+ * allow tests pass thanks to retries.
+ */
+ private static final String NODE_BOOTSTRAP_CFG = "{\n"
+ + " network: {\n"
+ + " port:{},\n"
+ + " nodeFinder:{\n"
+ + " netClusterNodes: [ {} ]\n"
+ + " }\n"
+ + " },\n"
+ + " raft.rpcInstallSnapshotTimeout: 10000"
+ + "}";
+
+ @WorkDirectory
+ private Path workDir;
+
+ private Cluster cluster;
+
+ @BeforeEach
+ void createCluster(TestInfo testInfo) {
+ cluster = new Cluster(testInfo, workDir, NODE_BOOTSTRAP_CFG);
+ }
+
+ @AfterEach
+ @Timeout(60)
+ void shutdownCluster() {
+ cluster.shutdown();
+ }
+
+ @Test
+ @Disabled
+ void assingmentsChangingOnNodeLeaveNodeJoin() throws Exception {
+ cluster.startAndInit(4);
+
+ createTestTable();
+
+ assertTrue(waitAssingments(List.of(
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2),
+ Set.of(0, 1, 2)
+ )));
+
+ TableImpl table = (TableImpl) cluster.node(0).tables().table("TEST");
+
+ BinaryRowEx key = new TupleMarshallerImpl(table.schemaView()).marshal(Tuple.create().set("id", 1));
+
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(0).node()).get());
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(1).node()).get());
+ assertNull(table.internalTable().get(key, new HybridClockImpl().now(), cluster.node(2).node()).get());
+
+ putData();
Review Comment:
It's a bit confusing that on the one hand we expose the key in order to retrieve the data and on the other hand encapsulate it into putData(). Lets be consistent ;)
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org