You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sa...@apache.org on 2024/01/25 15:04:23 UTC
(ignite-3) branch main updated: IGNITE-20993 Make the tables recover on the same assignments on different nodes (#3011)
This is an automated email from the ASF dual-hosted git repository.
sanpwc 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 23b25e666c IGNITE-20993 Make the tables recover on the same assignments on different nodes (#3011)
23b25e666c is described below
commit 23b25e666cfe972ae231239e4ed0c4fb4e0fc3e8
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Thu Jan 25 18:04:18 2024 +0300
IGNITE-20993 Make the tables recover on the same assignments on different nodes (#3011)
---
...niteDistributionZoneManagerNodeRestartTest.java | 2 +-
.../distributionzones/rebalance/RebalanceUtil.java | 28 +-
.../DistributionZoneRebalanceEngineTest.java | 33 +-
.../platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs | 32 +-
.../runner/app/ItIgniteNodeRestartTest.java | 499 +++++++++++++++++++--
.../ignite/internal/BaseIgniteRestartTest.java | 47 +-
.../internal/test/WatchListenerInhibitor.java | 15 +-
.../internal/table/distributed/TableManager.java | 71 ++-
.../table/distributed/TableManagerTest.java | 4 +-
9 files changed, 652 insertions(+), 79 deletions(-)
diff --git a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
index b7e051673d..b12c55f5ba 100644
--- a/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
+++ b/modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java
@@ -283,10 +283,10 @@ public class ItIgniteDistributionZoneManagerNodeRestartTest extends BaseIgniteRe
}
PartialNode partialNode = partialNode(
+ name,
nodeCfgMgr,
clusterCfgMgr,
metastore,
- null,
components,
localConfigurationGenerator,
logicalTopology,
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
index 3e95302031..3db7459342 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceUtil.java
@@ -17,7 +17,7 @@
package org.apache.ignite.internal.distributionzones.rebalance;
-import static java.util.Collections.emptySet;
+import static java.util.Collections.emptyList;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.and;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.exists;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
@@ -27,11 +27,11 @@ import static org.apache.ignite.internal.metastorage.dsl.Operations.ops;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
import static org.apache.ignite.internal.metastorage.dsl.Operations.remove;
import static org.apache.ignite.internal.metastorage.dsl.Statements.iif;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -270,7 +270,9 @@ public class RebalanceUtil {
int finalPartId = partId;
futures[partId] = tableAssignmentsFut.thenCompose(tableAssignments ->
- updatePendingAssignmentsKeys(
+ // TODO https://issues.apache.org/jira/browse/IGNITE-19763 We should distinguish empty stable assignments on
+ // TODO node recovery in case of interrupted table creation, and moving from empty assignments to non-empty.
+ tableAssignments.isEmpty() ? nullCompletedFuture() : updatePendingAssignmentsKeys(
tableDescriptor,
replicaGrpId,
dataNodes,
@@ -278,7 +280,7 @@ public class RebalanceUtil {
storageRevision,
metaStorageManager,
finalPartId,
- tableAssignments.isEmpty() ? emptySet() : tableAssignments.get(finalPartId)
+ tableAssignments.get(finalPartId)
));
}
@@ -507,16 +509,26 @@ public class RebalanceUtil {
return metaStorageManager.getAll(partitionKeysToPartitionNumber.keySet())
.thenApply(entries -> {
if (entries.isEmpty()) {
- return Collections.emptyList();
+ return emptyList();
}
Set<Assignment>[] result = new Set[numberOfPartitions];
+ int numberOfMsPartitions = 0;
- for (var entry : entries.entrySet()) {
- result[partitionKeysToPartitionNumber.get(entry.getKey())] = ByteUtils.fromBytes(entry.getValue().value());
+ for (var mapEntry : entries.entrySet()) {
+ Entry entry = mapEntry.getValue();
+
+ if (!entry.empty() && !entry.tombstone()) {
+ result[partitionKeysToPartitionNumber.get(mapEntry.getKey())] = ByteUtils.fromBytes(entry.value());
+ numberOfMsPartitions++;
+ }
}
- return Arrays.asList(result);
+ assert numberOfMsPartitions == 0 || numberOfMsPartitions == numberOfPartitions
+ : "Invalid number of stable partition entries received from meta storage [received="
+ + numberOfMsPartitions + ", numberOfPartitions=" + numberOfPartitions + ", tableId=" + tableId + "].";
+
+ return numberOfMsPartitions == 0 ? emptyList() : Arrays.asList(result);
});
}
diff --git a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
index bb499fd575..d4db410b60 100644
--- a/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
+++ b/modules/distribution-zones/src/test/java/org/apache/ignite/internal/distributionzones/rebalance/DistributionZoneRebalanceEngineTest.java
@@ -33,7 +33,6 @@ import static org.apache.ignite.internal.table.TableTestUtils.getTableIdStrict;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
-import static org.apache.ignite.internal.util.CompletableFutures.emptyMapCompletedFuture;
import static org.apache.ignite.sql.ColumnType.STRING;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -59,6 +58,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
+import org.apache.ignite.internal.affinity.AffinityUtils;
import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.commands.ColumnParams;
@@ -218,8 +218,20 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
return ret;
});
- // stable partitions for tables are empty
- when(metaStorageManager.getAll(any())).thenReturn(emptyMapCompletedFuture());
+ // stable partitions for tables
+ lenient().doAnswer(invocation -> {
+ Set<ByteArray> keys = invocation.getArgument(0);
+
+ Map<ByteArray, Entry> result = new HashMap<>();
+
+ for (var k : keys) {
+ var v = keyValueStorage.get(k.bytes());
+
+ result.put(k, v);
+ }
+
+ return completedFuture(result);
+ }).when(metaStorageManager).getAll(any());
}
@AfterEach
@@ -509,6 +521,21 @@ public class DistributionZoneRebalanceEngineTest extends IgniteAbstractTest {
List.of(ColumnParams.builder().name("k1").type(STRING).length(100).build()),
List.of("k1")
);
+
+ var tableId = getTableId(tableName);
+ var zoneId = getZoneId(zoneName);
+
+ CatalogZoneDescriptor zoneDescriptor = catalogManager.zone(zoneId, catalogManager.latestCatalogVersion());
+
+ Set<String> initialDataNodes = Set.of("node0");
+ List<Set<Assignment>> initialAssignments =
+ AffinityUtils.calculateAssignments(initialDataNodes, zoneDescriptor.partitions(), zoneDescriptor.replicas());
+
+ for (int i = 0; i < initialAssignments.size(); i++) {
+ var stableAssignmentPartitionKey = stablePartAssignmentsKey(new TablePartitionId(tableId, i)).bytes();
+
+ keyValueStorage.put(stableAssignmentPartitionKey, toBytes(initialAssignments.get(i)), clock.now());
+ }
}
private int getZoneId(String zoneName) {
diff --git a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
index 74debb835d..422b1e62cd 100644
--- a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
@@ -1,19 +1,19 @@
-/*
- * 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.
- */
+/*
+ * 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.
+ */
// <auto-generated/>
namespace Apache.Ignite
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index 28260e99fa..ad8af35156 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -17,16 +17,23 @@
package org.apache.ignite.internal.runner.app;
+import static java.util.Collections.emptySet;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.alterZone;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
+import static org.apache.ignite.sql.ColumnType.INT32;
import static org.apache.ignite.utils.ClusterServiceTestUtils.defaultSerializationRegistry;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -35,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -48,14 +56,17 @@ import java.util.Objects;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.IntFunction;
import java.util.function.LongFunction;
import java.util.function.LongSupplier;
-import java.util.stream.Collectors;
+import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
import org.apache.ignite.InitParameters;
@@ -63,8 +74,13 @@ import org.apache.ignite.internal.BaseIgniteRestartTest;
import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.app.ThreadPoolsManager;
+import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogManagerImpl;
import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.AlterZoneCommand;
+import org.apache.ignite.internal.catalog.commands.AlterZoneCommandBuilder;
+import org.apache.ignite.internal.catalog.commands.ColumnParams;
+import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.MakeIndexAvailableEventParameters;
@@ -95,10 +111,15 @@ import org.apache.ignite.internal.index.IndexManager;
import org.apache.ignite.internal.lang.ByteArray;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.WatchEvent;
+import org.apache.ignite.internal.metastorage.WatchListener;
import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
+import org.apache.ignite.internal.metastorage.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
@@ -163,11 +184,13 @@ import org.apache.ignite.tx.TransactionException;
import org.awaitility.Awaitility;
import org.intellij.lang.annotations.Language;
import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
/**
@@ -206,18 +229,33 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
@InjectConfiguration
private CriticalWorkersConfiguration workersConfiguration;
+ /**
+ * Interceptor of {@link MetaStorageManager#invoke(Condition, Collection, Collection)}.
+ */
+ private Map<Integer, InvokeInterceptor> metaStorageInvokeInterceptorByNode = new ConcurrentHashMap<>();
+
+ /**
+ * Mocks the data nodes returned by {@link DistributionZoneManager#dataNodes(long, int, int)} method on different nodes.
+ */
+ private Map<Integer, Supplier<CompletableFuture<Set<String>>>> dataNodesMockByNode = new ConcurrentHashMap<>();
+
+ @BeforeEach
+ public void beforeTest() {
+ metaStorageInvokeInterceptorByNode.clear();
+ dataNodesMockByNode.clear();
+ partialNodes.clear();
+ }
+
/**
* Start some of Ignite components that are able to serve as Ignite node for test purposes.
*
* @param idx Node index.
* @param cfgString Configuration string or {@code null} to use the default configuration.
- * @param revisionCallback Callback on storage revision update.
* @return Partial node.
*/
private PartialNode startPartialNode(
int idx,
- @Nullable @Language("HOCON") String cfgString,
- @Nullable Consumer<Long> revisionCallback
+ @Nullable @Language("HOCON") String cfgString
) {
String name = testNodeName(testInfo, idx);
@@ -318,16 +356,33 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
raftGroupEventsClientListener
);
+ var metaStorage = new RocksDbKeyValueStorage(name, dir.resolve("metastorage"));
+
+ InvokeInterceptor metaStorageInvokeInterceptor = metaStorageInvokeInterceptorByNode.get(idx);
+
var metaStorageMgr = new MetaStorageManagerImpl(
clusterSvc,
cmgManager,
logicalTopologyService,
raftMgr,
- new RocksDbKeyValueStorage(name, dir.resolve("metastorage")),
+ metaStorage,
hybridClock,
topologyAwareRaftGroupServiceFactory,
metaStorageConfiguration
- );
+ ) {
+ @Override
+ public CompletableFuture<Boolean> invoke(Condition condition, Collection<Operation> success, Collection<Operation> failure) {
+ if (metaStorageInvokeInterceptor != null) {
+ var res = metaStorageInvokeInterceptor.invoke(condition, success, failure);
+
+ if (res != null) {
+ return completedFuture(res);
+ }
+ }
+
+ return super.invoke(condition, success, failure);
+ }
+ };
var cfgStorage = new DistributedConfigurationStorage(metaStorageMgr);
@@ -390,7 +445,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
DataStorageManager dataStorageManager = new DataStorageManager(
dataStorageModules.createStorageEngines(
name,
- clusterConfigRegistry,
+ nodeCfgMgr.configurationRegistry(),
storagePath,
null
)
@@ -411,13 +466,24 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
SchemaManager schemaManager = new SchemaManager(registry, catalogManager, metaStorageMgr);
+ var dataNodesMock = dataNodesMockByNode.get(idx);
+
DistributionZoneManager distributionZoneManager = new DistributionZoneManager(
name,
registry,
metaStorageMgr,
logicalTopologyService,
catalogManager
- );
+ ) {
+ @Override
+ public CompletableFuture<Set<String>> dataNodes(long causalityToken, int catalogVersion, int zoneId) {
+ if (dataNodesMock != null) {
+ return dataNodesMock.get();
+ }
+
+ return super.dataNodes(causalityToken, catalogVersion, zoneId);
+ }
+ };
var schemaSyncService = new SchemaSyncServiceImpl(metaStorageMgr.clusterTime(), delayDurationMsSupplier);
@@ -462,7 +528,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
tableManager,
schemaManager,
dataStorageManager,
- () -> dataStorageModules.collectSchemasFields(modules.distributed().polymorphicSchemaExtensions()),
+ () -> dataStorageModules.collectSchemasFields(modules.local().polymorphicSchemaExtensions()),
replicaSvc,
hybridClock,
schemaSyncService,
@@ -520,10 +586,10 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
}
PartialNode partialNode = partialNode(
+ name,
nodeCfgMgr,
clusterCfgMgr,
metaStorageMgr,
- revisionCallback,
components,
localConfigurationGenerator,
logicalTopology,
@@ -563,7 +629,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
List<CompletableFuture<Ignite>> futures = IntStream.range(0, amount)
.mapToObj(i -> startNodeAsync(i, null))
- .collect(Collectors.toList());
+ .collect(toList());
if (initNeeded) {
String nodeName = CLUSTER_NODES_NAMES.get(0);
@@ -582,7 +648,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
return (IgniteImpl) future.join();
})
- .collect(Collectors.toList());
+ .collect(toList());
}
/**
@@ -1018,7 +1084,8 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
"[\"localhost:" + DEFAULT_NODE_PORT + "\"]",
DEFAULT_CLIENT_PORT + 11,
DEFAULT_HTTP_PORT + 11,
- DEFAULT_HTTPS_PORT + 11
+ DEFAULT_HTTPS_PORT + 11,
+ "{}"
);
IgniteImpl node1 = startNode(1, cfgString);
@@ -1065,7 +1132,7 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
*/
@Test
@Disabled(value = "https://issues.apache.org/jira/browse/IGNITE-18919")
- public void testMetastorageStop() {
+ public void testMetastorageStop() throws NodeStoppingException {
int cfgGap = 4;
List<IgniteImpl> nodes = startNodes(3);
@@ -1084,22 +1151,8 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
PartialNode partialNode = startPartialNode(
nodes.size() - 1,
- configurationString(nodes.size() - 1),
- rev -> {
- log.info("Partially started node: applying revision: " + rev);
-
- if (rev == cfgGap / 2) {
- log.info("Stopping METASTORAGE");
-
- stopNode(0);
-
- log.info("Starting METASTORAGE");
-
- startNode(0);
-
- log.info("Restarted METASTORAGE");
- }
- }
+ configurationString(nodes.size() - 1)
+ // TODO IGNITE-18919 here the revision callback was removed, because meta storage recovery process was changed.
);
TableManager tableManager = findComponent(partialNode.startedComponents(), TableManager.class);
@@ -1249,6 +1302,388 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
}
}
+ /**
+ * Checks that after create table call there is the same value of stable assignments in every node's meta storage, and it was
+ * changed only once.
+ *
+ * @param populateStableAssignmentsBeforeTableCreation Whether to populate stable assignments before table creation.
+ * @param restart Whether to restart one of the nodes while creating the table.
+ * @throws InterruptedException If interrupted.
+ */
+ @ParameterizedTest
+ @CsvSource({
+ "true,true",
+ "true,false",
+ "false,true",
+ "false,false"
+ })
+ public void createTableCallOnMultipleNodesTest(boolean populateStableAssignmentsBeforeTableCreation, boolean restart)
+ throws InterruptedException {
+ int nodesCount = 3;
+
+ var nodes = startNodes(3);
+
+ var node = nodes.get(0);
+
+ Map<Integer, AtomicInteger> stableAssignmentsChangeCounters = new ConcurrentHashMap<>();
+ Map<Integer, AtomicBoolean> lateChangeFlag = new ConcurrentHashMap<>();
+
+ // Prefix that will be updated after the table creation.
+ String testPrefix = "testPrefix";
+
+ for (int i = 0; i < nodesCount; i++) {
+ stableAssignmentsChangeCounters.put(i, new AtomicInteger());
+ lateChangeFlag.put(i, new AtomicBoolean());
+
+ final int fi = i;
+ createWatchListener(
+ nodes.get(i).metaStorageManager(),
+ STABLE_ASSIGNMENTS_PREFIX,
+ e -> stableAssignmentsChangeCounters.get(fi).incrementAndGet()
+ );
+
+ createWatchListener(
+ nodes.get(i).metaStorageManager(),
+ testPrefix,
+ e -> lateChangeFlag.get(fi).set(true)
+ );
+ }
+
+ // Assume that the table id will always be 7 for the test table. There is an assertion below to check this is true.
+ int tableId = 7;
+
+ var partId = new TablePartitionId(tableId, 0);
+
+ // Populate the stable assignments before calling table create, if needed.
+ if (populateStableAssignmentsBeforeTableCreation) {
+ node.metaStorageManager().put(stablePartAssignmentsKey(partId), ByteUtils.toBytes(Set.of(Assignment.forPeer(node.name()))));
+
+ waitForCondition(() -> lateChangeFlag.values().stream().allMatch(AtomicBoolean::get), 5_000);
+
+ lateChangeFlag.values().forEach(v -> v.set(false));
+ }
+
+ String zoneName = "TEST_ZONE";
+
+ if (restart) {
+ stopNode(nodesCount - 2);
+ }
+
+ try (Session session = node.sql().createSession()) {
+ session.execute(null, String.format("CREATE ZONE IF NOT EXISTS %s WITH REPLICAS=%d, PARTITIONS=%d", zoneName, nodesCount, 1));
+
+ session.execute(null, "CREATE TABLE " + TABLE_NAME
+ + "(id INT PRIMARY KEY, name VARCHAR) WITH PRIMARY_ZONE='" + zoneName + "';");
+ }
+
+ assertEquals(tableId, tableId(node, TABLE_NAME));
+
+ node.metaStorageManager().put(new ByteArray(testPrefix.getBytes(StandardCharsets.UTF_8)), new byte[0]);
+
+ if (restart) {
+ IgniteImpl restartedNode = startNode(nodesCount - 2);
+ nodes.set(nodesCount - 2, restartedNode);
+ }
+
+ // Waiting for late prefix on all nodes.
+ waitForCondition(() -> lateChangeFlag.values().stream().allMatch(AtomicBoolean::get), 5_000);
+
+ var assignmentsKey = stablePartAssignmentsKey(partId).bytes();
+
+ Set<Assignment> expectedAssignments = getAssignmentsFromMetaStorage(node.metaStorageManager(), assignmentsKey);
+
+ // Checking that the stable assignments are same and were changed only once on every node.
+ for (int i = 0; i < nodesCount; i++) {
+ if (!restart) {
+ // TODO IGNITE-21194 return this check
+ assertEquals(1, stableAssignmentsChangeCounters.get(i).get(), "node index=" + i);
+ }
+
+ assertEquals(
+ expectedAssignments,
+ getAssignmentsFromMetaStorage(nodes.get(i).metaStorageManager(), assignmentsKey),
+ "node index=" + i
+ );
+ }
+ }
+
+ private void createWatchListener(MetaStorageManager metaStorageManager, String prefix, Consumer<WatchEvent> listener) {
+ metaStorageManager.registerPrefixWatch(
+ new ByteArray(prefix.getBytes(StandardCharsets.UTF_8)),
+ new WatchListener() {
+ @Override
+ public CompletableFuture<Void> onUpdate(WatchEvent event) {
+ listener.accept(event);
+
+ return nullCompletedFuture();
+ }
+
+ @Override
+ public void onError(Throwable e) {
+ log.error("Error in test watch listener", e);
+ }
+ }
+ );
+ }
+
+ /**
+ * Creates the table on a cluster of 3 nodes, delays the table start processing, stops 2 nodes before the assignments are applied
+ * to meta storage. The remaining node is hanging, so it doesn't write the updates to meta storage as well. Then the stopped nodes
+ * are restarted, and receive different value of data nodes from distribution zones manager. We check that the calculated assignments
+ * for the table are eventually equal on every node's local meta storages, and they are equal to expected - to the assignments of
+ * the first node that is able to initialize them during the table start.
+ *
+ * @param nodeThatWrittenAssignments The index of the restarted node the has written the calculated assignments successfully.
+ * @param nodeThatPicksUpAssignments The index of the restarted node that picks up assignments from meta storage.
+ * @throws Exception If failed.
+ */
+ @ParameterizedTest
+ @CsvSource({
+ "1,2",
+ "2,1"
+ })
+ public void tableRecoveryOnMultipleRestartingNodes(int nodeThatWrittenAssignments, int nodeThatPicksUpAssignments) throws Exception {
+ var node0 = startNode(0);
+
+ int idx1 = 1;
+ int idx2 = 2;
+
+ var node1 = startPartialNode(idx1, configurationString(idx1));
+ var node2 = startPartialNode(idx2, configurationString(idx2));
+
+ String tableName = "TEST";
+ String zoneName = "ZONE_TEST";
+
+ // Assume that the table id is always 7, there is an assertion below to ensure this.
+ int tableId = 7;
+
+ var assignmentsKey = stablePartAssignmentsKey(new TablePartitionId(tableId, 0));
+
+ var metaStorageInterceptorFut = new CompletableFuture<>();
+ var metaStorageInterceptorInnerFut = new CompletableFuture<>();
+
+ metaStorageInvokeInterceptorByNode.put(nodeThatPicksUpAssignments,
+ (cond, success, failure) -> {
+ if (checkMetaStorageInvoke(success, assignmentsKey)) {
+ metaStorageInterceptorInnerFut.complete(null);
+
+ metaStorageInterceptorFut.join();
+ }
+
+ return null;
+ }
+ );
+
+ MetaStorageManager msManager1 = findComponent(node1.startedComponents(), MetaStorageManager.class);
+ MetaStorageManager msManager2 = findComponent(node2.startedComponents(), MetaStorageManager.class);
+
+ WatchListenerInhibitor nodeInhibitor0 = WatchListenerInhibitor.metastorageEventsInhibitor(node0.metaStorageManager());
+ WatchListenerInhibitor nodeInhibitor1 = WatchListenerInhibitor.metastorageEventsInhibitor(msManager1);
+ WatchListenerInhibitor nodeInhibitor2 = WatchListenerInhibitor.metastorageEventsInhibitor(msManager2);
+
+ // Create table, all nodes are lagging.
+ try (Session session = node0.sql().createSession()) {
+ session.execute(null, String.format("CREATE ZONE IF NOT EXISTS %s WITH REPLICAS=%d, PARTITIONS=%d", zoneName, 2, 1));
+
+ nodeInhibitor0.startInhibit();
+ nodeInhibitor1.startInhibit();
+ nodeInhibitor2.startInhibit();
+
+ session.executeAsync(null, "CREATE TABLE " + tableName
+ + "(id INT PRIMARY KEY, name VARCHAR) WITH PRIMARY_ZONE='" + zoneName + "';");
+ }
+
+ // Stopping 2 of 3 nodes.
+ node1.stop();
+ node2.stop();
+
+ Set<String> dataNodesOnNode1 = Set.of(node0.name(), node1.name());
+ Set<String> dataNodesOnNode2 = Set.of(node1.name(), node2.name());
+
+ dataNodesMockByNode.put(idx1, () -> completedFuture(dataNodesOnNode1));
+ dataNodesMockByNode.put(idx2, () -> completedFuture(dataNodesOnNode2));
+
+ AtomicReference<PartialNode> nodeWnRef = new AtomicReference<>();
+ AtomicReference<PartialNode> nodePnRef = new AtomicReference<>();
+
+ // Restarting 2 nodes.
+ var nodePnFut = runAsync(() ->
+ nodePnRef.set(startPartialNode(nodeThatPicksUpAssignments, configurationString(nodeThatPicksUpAssignments))));
+
+ assertThat(metaStorageInterceptorInnerFut, willCompleteSuccessfully());
+
+ var nodeWnFut = runAsync(() ->
+ nodeWnRef.set(startPartialNode(nodeThatWrittenAssignments, configurationString(nodeThatWrittenAssignments))));
+
+ assertThat(nodeWnFut, willCompleteSuccessfully());
+
+ var msManagerRestartedW = findComponent(nodeWnRef.get().startedComponents(), MetaStorageManager.class);
+ waitForValueInLocalMs(msManagerRestartedW, assignmentsKey);
+
+ metaStorageInterceptorFut.complete(null);
+
+ assertThat(nodePnFut, willCompleteSuccessfully());
+
+ var msManagerRestartedP = findComponent(nodePnRef.get().startedComponents(), MetaStorageManager.class);
+
+ waitForValueInLocalMs(msManagerRestartedP, assignmentsKey);
+
+ // Node 0 stops hanging.
+ nodeInhibitor0.stopInhibit();
+ waitForValueInLocalMs(node0.metaStorageManager(), assignmentsKey);
+
+ assertEquals(tableId, tableId(node0, tableName));
+
+ Set<Assignment> expectedAssignments = dataNodesMockByNode.get(nodeThatWrittenAssignments).get().join()
+ .stream().map(Assignment::forPeer).collect(toSet());
+
+ checkAssignmentsInMetaStorage(node0.metaStorageManager(), assignmentsKey.bytes(), expectedAssignments);
+ checkAssignmentsInMetaStorage(msManagerRestartedW, assignmentsKey.bytes(), expectedAssignments);
+ checkAssignmentsInMetaStorage(msManagerRestartedP, assignmentsKey.bytes(), expectedAssignments);
+ }
+
+ @Test
+ public void testSequentialAsyncTableCreationThenAlterZoneThenRestartOnMsSnapshot() throws InterruptedException {
+ var node0 = startNode(0);
+ var node1 = startNode(1);
+
+ String tableName = "TEST";
+ String zoneName = "ZONE_TEST";
+
+ try (Session session = node0.sql().createSession()) {
+ session.execute(null, String.format("CREATE ZONE IF NOT EXISTS %s WITH REPLICAS=%d, PARTITIONS=%d", zoneName, 2, 1));
+ }
+
+ int catalogVersionBeforeTable = node0.catalogManager().latestCatalogVersion();
+
+ WatchListenerInhibitor nodeInhibitor0 = WatchListenerInhibitor.metastorageEventsInhibitor(node0);
+ WatchListenerInhibitor nodeInhibitor1 = WatchListenerInhibitor.metastorageEventsInhibitor(node1);
+
+ nodeInhibitor0.startInhibit();
+ nodeInhibitor1.startInhibit();
+
+ // Assume that the table id is always 7, there is an assertion below to ensure this.
+ int tableId = 7;
+
+ var assignmentsKey = stablePartAssignmentsKey(new TablePartitionId(tableId, 0));
+
+ var tableFut = createTableInCatalog(node0.catalogManager(), tableName, zoneName);
+
+ stopNode(1);
+
+ var alterZoneFut = alterZoneAsync(node0.catalogManager(), zoneName, 1);
+
+ // Wait for the next catalog version: table creation.
+ // The next catalog update (alter zone) can't be processed until the table creation is completed.
+ assertTrue(waitForCondition(() -> {
+ int ver = latestCatalogVersionInMs(node0.metaStorageManager());
+
+ return ver == catalogVersionBeforeTable + 1;
+ }, 10_000));
+
+ // Double check that the version is exactly as expected, if it's not, the synchronous sequential catalog update processing may be
+ // broken.
+ assertEquals(catalogVersionBeforeTable + 1, latestCatalogVersionInMs(node0.metaStorageManager()));
+
+ Thread.sleep(1000);
+
+ forceSnapshotUsageOnRestart(node0);
+
+ log.info("Restarting node 1.");
+
+ node1 = startNode(1);
+
+ waitForValueInLocalMs(node1.metaStorageManager(), assignmentsKey);
+
+ nodeInhibitor0.stopInhibit();
+
+ assertThat(tableFut, willCompleteSuccessfully());
+ assertThat(alterZoneFut, willCompleteSuccessfully());
+
+ assertEquals(tableId, tableId(node0, tableName));
+
+ waitForValueInLocalMs(node0.metaStorageManager(), assignmentsKey);
+
+ var finalNode1 = node1;
+
+ // Restart is followed by rebalance, because data nodes are recalculated after full table creation that is completed after restart.
+ assertTrue(waitForCondition(() -> {
+ Set<Assignment> assignments0 = getAssignmentsFromMetaStorage(node0.metaStorageManager(), assignmentsKey.bytes());
+ Set<Assignment> assignments1 = getAssignmentsFromMetaStorage(finalNode1.metaStorageManager(), assignmentsKey.bytes());
+
+ return assignments0.size() == 1 && assignments0.equals(assignments1);
+
+ }, 10_000));
+ }
+
+ private int latestCatalogVersionInMs(MetaStorageManager metaStorageManager) {
+ var e = metaStorageManager.getLocally(new ByteArray("catalog.version".getBytes(StandardCharsets.UTF_8)), 1000);
+
+ if (e == null || e.empty()) {
+ return -1;
+ }
+
+ return ByteUtils.bytesToInt(e.value());
+ }
+
+ private static CompletableFuture<Void> createTableInCatalog(CatalogManager catalogManager, String tableName, String zoneName) {
+ var tableColumn = ColumnParams.builder().name("id").type(INT32).build();
+
+ var createTableCommand = CreateTableCommand.builder()
+ .schemaName("PUBLIC")
+ .tableName(tableName)
+ .columns(List.of(tableColumn))
+ .primaryKeyColumns(List.of("id"))
+ .zone(zoneName)
+ .build();
+
+ return catalogManager.execute(createTableCommand);
+ }
+
+ private static CompletableFuture<Void> alterZoneAsync(
+ CatalogManager catalogManager,
+ String zoneName,
+ @Nullable Integer replicas
+ ) {
+ AlterZoneCommandBuilder builder = AlterZoneCommand.builder().zoneName(zoneName);
+
+ builder.replicas(replicas);
+
+ return catalogManager.execute(builder.build());
+ }
+
+ private void waitForValueInLocalMs(MetaStorageManager metaStorageManager, ByteArray key) throws InterruptedException {
+ assertTrue(waitForCondition(() -> {
+ var e = metaStorageManager.getLocally(key, metaStorageManager.appliedRevision());
+
+ return !e.empty();
+ }, 10_000));
+ }
+
+ private boolean checkMetaStorageInvoke(Collection<Operation> ops, ByteArray key) {
+ var k = new String(key.bytes(), StandardCharsets.UTF_8);
+
+ return ops.stream().anyMatch(op -> new String(op.key(), StandardCharsets.UTF_8).equals(k));
+ }
+
+ private void checkAssignmentsInMetaStorage(MetaStorageManager metaStorageManager, byte[] assignmentsKey, Set<Assignment> expected) {
+ Set<Assignment> a = getAssignmentsFromMetaStorage(metaStorageManager, assignmentsKey);
+
+ assertEquals(expected, a);
+ }
+
+ private Set<Assignment> getAssignmentsFromMetaStorage(MetaStorageManager metaStorageManager, byte[] assignmentsKey) {
+ var e = metaStorageManager.getLocally(new ByteArray(assignmentsKey), metaStorageManager.appliedRevision());
+
+ return e == null || e.tombstone() || e.empty()
+ ? emptySet()
+ : ByteUtils.fromBytes(e.value());
+ }
+
+ private int tableId(Ignite node, String tableName) {
+ return ((TableImpl) node.tables().table(tableName)).tableId();
+ }
+
/**
* Checks the table exists and validates all data in it.
*
@@ -1364,4 +1799,8 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
return ignite.tables().table(name);
}
+
+ private interface InvokeInterceptor {
+ Boolean invoke(Condition condition, Collection<Operation> success, Collection<Operation> failure);
+ }
}
diff --git a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/BaseIgniteRestartTest.java b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/BaseIgniteRestartTest.java
index 317fe3628f..1afc9df950 100644
--- a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/BaseIgniteRestartTest.java
+++ b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/BaseIgniteRestartTest.java
@@ -31,7 +31,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
+import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgnitionManager;
import org.apache.ignite.InitParameters;
@@ -84,6 +84,7 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
+ "}";
/** Nodes bootstrap configuration pattern. */
+ @Language("HOCON")
protected static final String NODE_BOOTSTRAP_CFG = "{\n"
+ " network.port: {},\n"
+ " network.nodeFinder.netClusterNodes: {}\n"
@@ -101,7 +102,10 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
+ " rest: {\n"
+ " port: {}, \n"
+ " ssl.port: {} \n"
- + " }\n"
+ + " },\n"
+ + " nodeAttributes: {"
+ + " nodeAttributes: {}"
+ + " }"
+ "}";
public TestInfo testInfo;
@@ -116,7 +120,7 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
@BeforeEach
void setUp(TestInfo testInfo) {
this.testInfo = testInfo;
- this.partialNodes = new ArrayList<>();
+ this.partialNodes = new CopyOnWriteArrayList<>();
}
/**
@@ -215,6 +219,17 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
* @return Configuration string.
*/
protected static String configurationString(int idx) {
+ return configurationString(idx, "{}");
+ }
+
+ /**
+ * Build a configuration string.
+ *
+ * @param idx Node index.
+ * @param attributes Node attributes, should be empty string if not needed.
+ * @return Configuration string.
+ */
+ protected static String configurationString(int idx, String attributes) {
int port = DEFAULT_NODE_PORT + idx;
int clientPort = DEFAULT_CLIENT_PORT + idx;
int httpPort = DEFAULT_HTTP_PORT + idx;
@@ -223,16 +238,16 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
// The address of the first node.
@Language("HOCON") String connectAddr = "[localhost\":\"" + DEFAULT_NODE_PORT + "]";
- return IgniteStringFormatter.format(NODE_BOOTSTRAP_CFG, port, connectAddr, clientPort, httpPort, httpsPort);
+ return IgniteStringFormatter.format(NODE_BOOTSTRAP_CFG, port, connectAddr, clientPort, httpPort, httpsPort, attributes);
}
/**
* Returns partial node. Chains deploying watches to configuration notifications and waits for it,
* so returned partial node is started and ready to work.
*
+ * @param name Node name.
* @param nodeCfgMgr Node configuration manager.
* @param clusterCfgMgr Cluster configuration manager.
- * @param revisionCallback RevisionCallback Callback on storage revision update.
* @param components Started components of a node.
* @param localConfigurationGenerator Local configuration generator.
* @param logicalTopology Logical topology.
@@ -242,10 +257,10 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
* @return Partial node.
*/
public PartialNode partialNode(
+ String name,
ConfigurationManager nodeCfgMgr,
ConfigurationManager clusterCfgMgr,
MetaStorageManager metaStorageMgr,
- @Nullable Consumer<Long> revisionCallback,
List<IgniteComponent> components,
ConfigurationTreeGenerator localConfigurationGenerator,
LogicalTopologyImpl logicalTopology,
@@ -272,6 +287,7 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
log.info("Completed recovery on partially started node, MetaStorage revision recovered to: " + recoveryRevision);
return new PartialNode(
+ name,
components,
List.of(localConfigurationGenerator, distributedConfigurationGenerator),
logicalTopology,
@@ -360,6 +376,8 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
* Node with partially started components.
*/
public static class PartialNode {
+ private final String name;
+
private final List<IgniteComponent> startedComponents;
private final List<ManuallyCloseable> closeables;
@@ -371,12 +389,14 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
private final HybridClock clock;
PartialNode(
+ String name,
List<IgniteComponent> startedComponents,
List<ManuallyCloseable> closeables,
LogicalTopology logicalTopology,
IgniteLogger log,
HybridClock clock
) {
+ this.name = name;
this.startedComponents = startedComponents;
this.closeables = closeables;
this.logicalTopology = logicalTopology;
@@ -384,6 +404,15 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
this.clock = clock;
}
+ /**
+ * Node name.
+ *
+ * @return Node name.
+ */
+ public String name() {
+ return name;
+ }
+
/**
* Stops node.
*/
@@ -393,7 +422,11 @@ public abstract class BaseIgniteRestartTest extends IgniteAbstractTest {
while (iter.hasPrevious()) {
IgniteComponent prev = iter.previous();
- prev.beforeNodeStop();
+ try {
+ prev.beforeNodeStop();
+ } catch (Exception e) {
+ log.error("Error during calling `beforeNodeStop`", e);
+ }
}
iter = startedComponents.listIterator(startedComponents.size());
diff --git a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
index 8e5b5792d1..b6532e47cb 100644
--- a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
+++ b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/test/WatchListenerInhibitor.java
@@ -25,6 +25,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.ReadWriteLock;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import org.apache.ignite.internal.metastorage.server.WatchProcessor;
import org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
@@ -53,10 +54,20 @@ public class WatchListenerInhibitor {
public static WatchListenerInhibitor metastorageEventsInhibitor(Ignite ignite) {
IgniteImpl igniteImpl = (IgniteImpl) ignite;
- var metaStorageManager = (MetaStorageManagerImpl) igniteImpl.metaStorageManager();
+ return metastorageEventsInhibitor(igniteImpl.metaStorageManager());
+ }
+
+ /**
+ * Creates the specific listener which can inhibit events for real metastorage listener.
+ *
+ * @param metaStorageManager Meta storage manager.
+ * @return Listener inhibitor.
+ */
+ public static WatchListenerInhibitor metastorageEventsInhibitor(MetaStorageManager metaStorageManager) {
+ var metaStorageManager0 = metaStorageManager;
//TODO: IGNITE-15723 After a component factory is implemented, need to got rid of reflection here.
- var storage = (RocksDbKeyValueStorage) getFieldValue(metaStorageManager, MetaStorageManagerImpl.class, "storage");
+ var storage = (RocksDbKeyValueStorage) getFieldValue(metaStorageManager0, MetaStorageManagerImpl.class, "storage");
var watchProcessor = (WatchProcessor) getFieldValue(storage, RocksDbKeyValueStorage.class, "watchProcessor");
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 911aa1ff21..b1a9e4b084 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -26,6 +26,7 @@ import static java.util.concurrent.CompletableFuture.anyOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.concurrent.CompletableFuture.runAsync;
+import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.causality.IncrementalVersionedValue.dependingOn;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.partitionAssignmentsGetLocally;
import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.tableAssignmentsGetLocally;
@@ -74,7 +75,6 @@ import java.util.function.Function;
import java.util.function.IntSupplier;
import java.util.function.LongFunction;
import java.util.function.Supplier;
-import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.ignite.internal.affinity.AffinityUtils;
@@ -609,7 +609,15 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return createTableLocally(parameters.causalityToken(), parameters.catalogVersion(), parameters.tableDescriptor(), false);
}
- private CompletableFuture<Boolean> writeTableAssignmentsToMetastore(
+ /**
+ * Writes the set of assignments to meta storage. If there are some assignments already, gets them from meta storage. Returns
+ * the list of assignments that really are in meta storage.
+ *
+ * @param tableId Table id.
+ * @param assignmentsFuture Assignments future, to get the assignments that should be written.
+ * @return Real list of assignments.
+ */
+ private CompletableFuture<List<Set<Assignment>>> writeTableAssignmentsToMetastore(
int tableId,
CompletableFuture<List<Set<Assignment>>> assignmentsFuture
) {
@@ -629,6 +637,41 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return metaStorageMgr
.invoke(condition, partitionAssignments, Collections.emptyList())
+ .thenCompose(invokeResult -> {
+ if (invokeResult) {
+ LOG.info(IgniteStringFormatter.format("Assignments calculated from data nodes are successfully "
+ + "written to meta storage [tableId={}, assignments={}]", tableId, newAssignments));
+
+ return completedFuture(newAssignments);
+ } else {
+ Set<ByteArray> partKeys = IntStream.range(0, newAssignments.size())
+ .mapToObj(p -> stablePartAssignmentsKey(new TablePartitionId(tableId, p)))
+ .collect(toSet());
+
+ CompletableFuture<Map<ByteArray, Entry>> resFuture = metaStorageMgr.getAll(partKeys);
+
+ return resFuture.thenApply(metaStorageAssignments -> {
+ List<Set<Assignment>> realAssignments = new ArrayList<>();
+
+ for (int p = 0; p < newAssignments.size(); p++) {
+ var partId = new TablePartitionId(tableId, p);
+ Entry assignmentsEntry = metaStorageAssignments.get(stablePartAssignmentsKey(partId));
+
+ assert assignmentsEntry != null && !assignmentsEntry.empty() && !assignmentsEntry.tombstone()
+ : "Unexpected assignments for partition [" + partId + ", entry=" + assignmentsEntry + "].";
+
+ Set<Assignment> real = ByteUtils.fromBytes(assignmentsEntry.value());
+
+ realAssignments.add(real);
+ }
+
+ LOG.info(IgniteStringFormatter.format("Assignments picked up from meta storage [tableId={}, "
+ + "assignments={}]", tableId, realAssignments));
+
+ return realAssignments;
+ });
+ }
+ })
.exceptionally(e -> {
LOG.error("Couldn't write assignments to metastore", e);
@@ -1095,9 +1138,9 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
CompletableFuture<List<Set<Assignment>>> assignmentsFuture;
- // Check if the table already has assignments in the vault.
- // So, it means, that it is a recovery process and we should use the vault assignments instead of calculation for the new ones.
- // TODO https://issues.apache.org/jira/browse/IGNITE-20993
+ // Check if the table already has assignments in the meta storage locally.
+ // So, it means, that it is a recovery process and we should use the meta storage local assignments instead of calculation
+ // of the new ones.
if (partitionAssignmentsGetLocally(metaStorageMgr, tableId, 0, causalityToken) != null) {
assignmentsFuture = completedFuture(
tableAssignmentsGetLocally(metaStorageMgr, tableId, zoneDescriptor.partitions(), causalityToken));
@@ -1108,16 +1151,24 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
zoneDescriptor.partitions(),
zoneDescriptor.replicas()
));
+
+ assignmentsFuture.thenAccept(assignmentsList -> {
+ LOG.info(IgniteStringFormatter.format("Assignments calculated from data nodes [table={}, tableId={}, assignments={}, "
+ + "revision={}]", tableDescriptor.name(), tableId, assignmentsList, causalityToken));
+ });
}
+ CompletableFuture<List<Set<Assignment>>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(tableId, assignmentsFuture);
+
return createTableLocally(
causalityToken,
tableDescriptor,
zoneDescriptor,
- assignmentsFuture,
+ assignmentsFutureAfterInvoke,
catalogVersion,
onNodeRecovery
- ).thenCompose(ignored -> writeTableAssignmentsToMetastore(tableId, assignmentsFuture));
+ );
});
}
@@ -1336,7 +1387,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
Set<ByteArray> assignmentKeys = IntStream.range(0, partitions)
.mapToObj(p -> stablePartAssignmentsKey(new TablePartitionId(tableId, p)))
- .collect(Collectors.toSet());
+ .collect(toSet());
metaStorageMgr.removeAll(assignmentKeys);
}
@@ -1684,8 +1735,8 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
var stringKey = new String(pendingAssignmentsEntry.key(), UTF_8);
LOG.info("Received update on pending assignments. Check if new raft group should be started"
- + " [key={}, partition={}, table={}, localMemberAddress={}]",
- stringKey, partId, table.name(), localNode().address());
+ + " [key={}, partition={}, table={}, localMemberAddress={}, pendingAssignments={}]",
+ stringKey, partId, table.name(), localNode().address(), pendingAssignments);
}
Set<Assignment> stableAssignments = stableAssignmentsEntry.value() == null
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
index 8afa0cb9bb..984c7c50f8 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
@@ -597,8 +597,8 @@ public class TableManagerTest extends IgniteAbstractTest {
subscriber.onComplete();
});
- when(msm.invoke(any(), any(Operation.class), any(Operation.class))).thenReturn(nullCompletedFuture());
- when(msm.invoke(any(), any(List.class), any(List.class))).thenReturn(nullCompletedFuture());
+ when(msm.invoke(any(), any(Operation.class), any(Operation.class))).thenReturn(trueCompletedFuture());
+ when(msm.invoke(any(), any(List.class), any(List.class))).thenReturn(trueCompletedFuture());
when(msm.get(any())).thenReturn(nullCompletedFuture());
when(msm.recoveryFinishedFuture()).thenReturn(completedFuture(1L));