You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2023/08/17 07:53:30 UTC
[ignite-3] branch main updated: IGNITE-20229 Get rid of configurations in PlacementDriverManager (#2453)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill 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 1e9fe13003 IGNITE-20229 Get rid of configurations in PlacementDriverManager (#2453)
1e9fe13003 is described below
commit 1e9fe13003bda48caf5804657bfa7ca29b42eece
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Thu Aug 17 10:53:25 2023 +0300
IGNITE-20229 Get rid of configurations in PlacementDriverManager (#2453)
---
.../apache/ignite/internal/util/IgniteUtils.java | 36 ++++-
modules/placement-driver/build.gradle | 1 +
.../internal/placementdriver/ActiveActorTest.java | 14 +-
.../placementdriver/BasePlacementDriverTest.java | 75 +++++++++
.../MultiActorPlacementDriverTest.java | 70 +--------
.../PlacementDriverManagerTest.java | 69 +--------
.../internal/placementdriver/LeaseUpdater.java | 106 +++++--------
.../placementdriver/PlacementDriverManager.java | 104 +++++--------
.../internal/placementdriver/TopologyTracker.java | 3 +-
.../placementdriver/leases/LeaseTracker.java | 168 ++++++++++-----------
.../internal/placementdriver/leases/Leases.java | 43 ++++++
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
12 files changed, 324 insertions(+), 368 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 3d2964b442..94fa38aba1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.util;
import static java.nio.file.StandardOpenOption.CREATE;
import static java.nio.file.StandardOpenOption.WRITE;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
import java.io.IOException;
@@ -880,10 +881,10 @@ public class IgniteUtils {
/**
* Method that runs the provided {@code fn} in {@code busyLock}.
*
- * @param busyLock Component's busy lock
- * @param fn Function to run
- * @param <T> Type of returned value from {@code fn}
- * @return Result of the provided function
+ * @param busyLock Component's busy lock.
+ * @param fn Function to run.
+ * @param <T> Type of returned value from {@code fn}.
+ * @return Result of the provided function.
*/
public static <T> T inBusyLock(IgniteSpinBusyLock busyLock, Supplier<T> fn) {
if (!busyLock.enterBusy()) {
@@ -899,8 +900,8 @@ public class IgniteUtils {
/**
* Method that runs the provided {@code fn} in {@code busyLock}.
*
- * @param busyLock Component's busy lock
- * @param fn Runnable to run
+ * @param busyLock Component's busy lock.
+ * @param fn Runnable to run.
*/
public static void inBusyLock(IgniteSpinBusyLock busyLock, Runnable fn) {
if (!busyLock.enterBusy()) {
@@ -913,6 +914,29 @@ public class IgniteUtils {
}
}
+ /**
+ * Method that runs the provided {@code fn} in {@code busyLock}.
+ *
+ * @param <T> Type of returned value from {@code fn}.
+ * @param busyLock Component's busy lock.
+ * @param fn Function to run.
+ * @return Future returned from the {@code fn}, or future with the {@link NodeStoppingException} if
+ * {@link IgniteSpinBusyLock#enterBusy()} failed or with runtime exception/error while executing the {@code fn}.
+ */
+ public static <T> CompletableFuture<T> inBusyLockAsync(IgniteSpinBusyLock busyLock, Supplier<CompletableFuture<T>> fn) {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ return fn.get();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
/**
* Collects all the fields of given class which are defined as a public static within the specified class.
*
diff --git a/modules/placement-driver/build.gradle b/modules/placement-driver/build.gradle
index e939fe2bd3..4de8e53528 100644
--- a/modules/placement-driver/build.gradle
+++ b/modules/placement-driver/build.gradle
@@ -42,6 +42,7 @@ dependencies {
testFixturesImplementation project(':ignite-core')
+ integrationTestImplementation libs.jetbrains.annotations
integrationTestImplementation project(':ignite-network')
integrationTestImplementation project(':ignite-raft')
integrationTestImplementation project(':ignite-raft-api')
diff --git a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
index 6ef7b70be1..803928c120 100644
--- a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
+++ b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
@@ -52,7 +52,6 @@ import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopolog
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.dsl.Operation;
@@ -70,9 +69,9 @@ import org.apache.ignite.internal.raft.server.impl.JraftServerImpl;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
import org.apache.ignite.network.ClusterNode;
@@ -116,12 +115,6 @@ public class ActiveActorTest extends IgniteAbstractTest {
@Mock
MetaStorageManager msm;
- @InjectConfiguration
- private TablesConfiguration tblsCfg;
-
- @InjectConfiguration
- private DistributionZonesConfiguration dstZnsCfg;
-
@AfterEach
@Override
public void tearDown(TestInfo testInfo) throws Exception {
@@ -131,6 +124,8 @@ public class ActiveActorTest extends IgniteAbstractTest {
placementDriverManagers.clear();
+ IgniteUtils.shutdownAndAwaitTermination(executor, 10, TimeUnit.SECONDS);
+
super.tearDown(testInfo);
}
@@ -162,6 +157,7 @@ public class ActiveActorTest extends IgniteAbstractTest {
);
PlacementDriverManager placementDriverManager = new PlacementDriverManager(
+ nodeName,
msm,
new VaultManager(new InMemoryVaultService()),
GROUP_ID,
@@ -170,8 +166,6 @@ public class ActiveActorTest extends IgniteAbstractTest {
logicalTopologyService,
raftManager,
raftGroupServiceFactory,
- tblsCfg,
- dstZnsCfg,
new HybridClockImpl()
);
diff --git a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/BasePlacementDriverTest.java b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/BasePlacementDriverTest.java
new file mode 100644
index 0000000000..707e84f80e
--- /dev/null
+++ b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/BasePlacementDriverTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.placementdriver;
+
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static org.apache.ignite.internal.utils.RebalanceUtil.stablePartAssignmentsKey;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.placementdriver.leases.Lease;
+import org.apache.ignite.internal.placementdriver.leases.LeaseBatch;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.apache.ignite.internal.util.ByteUtils;
+import org.apache.ignite.lang.ByteArray;
+import org.jetbrains.annotations.Nullable;
+
+/** Base class for testing the placement driver. */
+abstract class BasePlacementDriverTest extends IgniteAbstractTest {
+ /**
+ * Creates an assignment for the fake table.
+ *
+ * @return Replication group id.
+ */
+ protected static TablePartitionId createTableAssignment(MetaStorageManager metastore, int tableId, List<String> dataNodes) {
+ List<Set<Assignment>> assignments = AffinityUtils.calculateAssignments(dataNodes, 1, dataNodes.size());
+
+ Map<ByteArray, byte[]> partitionAssignments = new HashMap<>(assignments.size());
+
+ for (int i = 0; i < assignments.size(); i++) {
+ partitionAssignments.put(
+ stablePartAssignmentsKey(new TablePartitionId(tableId, i)),
+ ByteUtils.toBytes(assignments.get(i)));
+ }
+
+ metastore.putAll(partitionAssignments).join();
+
+ var grpPart0 = new TablePartitionId(tableId, 0);
+
+ log.info("Fake table created [id={}, repGrp={}]", tableId, grpPart0);
+
+ return grpPart0;
+ }
+
+ protected static @Nullable Lease leaseFromBytes(byte[] bytes, ReplicationGroupId groupId) {
+ LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(bytes).order(LITTLE_ENDIAN));
+
+ return leaseBatch.leases().stream()
+ .filter(l -> l.replicationGroupId().equals(groupId))
+ .findAny()
+ .orElse(null);
+ }
+}
diff --git a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/MultiActorPlacementDriverTest.java b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/MultiActorPlacementDriverTest.java
index a00dbba7da..aafcda5cfa 100644
--- a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/MultiActorPlacementDriverTest.java
+++ b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/MultiActorPlacementDriverTest.java
@@ -18,12 +18,10 @@
package org.apache.ignite.internal.placementdriver;
import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.getZoneId;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_LEASES_KEY;
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.utils.RebalanceUtil.stablePartAssignmentsKey;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -31,26 +29,20 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.Closeable;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import org.apache.ignite.internal.affinity.AffinityUtils;
-import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
@@ -59,7 +51,6 @@ import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStora
import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
import org.apache.ignite.internal.placementdriver.PlacementDriverManagerTest.LogicalTopologyServiceTestImpl;
import org.apache.ignite.internal.placementdriver.leases.Lease;
-import org.apache.ignite.internal.placementdriver.leases.LeaseBatch;
import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessageGroup;
@@ -68,17 +59,11 @@ import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplica
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupServiceFactory;
import org.apache.ignite.internal.raft.configuration.RaftConfiguration;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
-import org.apache.ignite.internal.schema.configuration.ExtendedTableChange;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
-import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.inmemory.InMemoryVaultService;
-import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteTriFunction;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.NetworkAddress;
@@ -97,7 +82,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
* There are tests of muti-nodes for placement driver.
*/
@ExtendWith(ConfigurationExtension.class)
-public class MultiActorPlacementDriverTest extends IgniteAbstractTest {
+public class MultiActorPlacementDriverTest extends BasePlacementDriverTest {
public static final int BASE_PORT = 1234;
private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();
@@ -107,12 +92,6 @@ public class MultiActorPlacementDriverTest extends IgniteAbstractTest {
@InjectConfiguration
private RaftConfiguration raftConfiguration;
- @InjectConfiguration
- private TablesConfiguration tblsCfg;
-
- @InjectConfiguration("mock.distributionZones {zone1 = { partitions = 1, replicas = 2, zoneId = 1}}")
- private DistributionZonesConfiguration dstZnsCfg;
-
@InjectConfiguration
private MetaStorageConfiguration metaStorageConfiguration;
@@ -297,16 +276,15 @@ public class MultiActorPlacementDriverTest extends IgniteAbstractTest {
}
var placementDriverManager = new PlacementDriverManager(
+ nodeName,
metaStorageManager,
vaultManager,
MetastorageGroupId.INSTANCE,
clusterService,
- () -> cmgManager.metaStorageNodes(),
+ cmgManager::metaStorageNodes,
logicalTopologyService,
raftManager,
topologyAwareRaftGroupServiceFactory,
- tblsCfg,
- dstZnsCfg,
clock
);
@@ -596,46 +574,8 @@ public class MultiActorPlacementDriverTest extends IgniteAbstractTest {
* Creates an assignment for the fake table.
*
* @return Replication group id.
- * @throws Exception If failed.
*/
- private TablePartitionId createTableAssignment() throws Exception {
- int tableId = nextTableId.incrementAndGet();
-
- List<Set<Assignment>> assignments = AffinityUtils.calculateAssignments(nodeNames, 1, nodeNames.size());
-
- tblsCfg.tables().change(tableViewTableChangeNamedListChange -> {
- tableViewTableChangeNamedListChange.create("test-table-" + tableId, tableChange -> {
- var extConfCh = ((ExtendedTableChange) tableChange);
-
- extConfCh.changeId(tableId);
-
- extConfCh.changeZoneId(getZoneId(dstZnsCfg, "zone1"));
- });
- }).thenCompose(v -> {
- Map<ByteArray, byte[]> partitionAssignments = new HashMap<>(assignments.size());
-
- for (int i = 0; i < assignments.size(); i++) {
- partitionAssignments.put(
- stablePartAssignmentsKey(
- new TablePartitionId(tableId, i)),
- ByteUtils.toBytes(assignments.get(i)));
-
- }
-
- return metaStorageManager.putAll(partitionAssignments);
- })
- .get();
-
- var grpPart0 = new TablePartitionId(tableId, 0);
-
- log.info("Fake table created [id={}, repGrp={}]", tableId, grpPart0);
-
- return grpPart0;
- }
-
- private Lease leaseFromBytes(byte[] bytes, ReplicationGroupId groupId) {
- LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN));
-
- return leaseBatch.leases().stream().filter(l -> l.replicationGroupId().equals(groupId)).findAny().orElse(null);
+ private TablePartitionId createTableAssignment() {
+ return createTableAssignment(metaStorageManager, nextTableId.get(), nodeNames);
}
}
diff --git a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/PlacementDriverManagerTest.java b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/PlacementDriverManagerTest.java
index afcad20b6a..38dc455d7c 100644
--- a/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/PlacementDriverManagerTest.java
+++ b/modules/placement-driver/src/integrationTest/java/org/apache/ignite/internal/placementdriver/PlacementDriverManagerTest.java
@@ -20,13 +20,11 @@ package org.apache.ignite.internal.placementdriver;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.internal.affinity.AffinityUtils.calculateAssignmentForPartition;
-import static org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil.getZoneId;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_LEASES_KEY;
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.utils.RebalanceUtil.STABLE_ASSIGNMENTS_PREFIX;
-import static org.apache.ignite.internal.utils.RebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.lang.ByteArray.fromString;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -36,20 +34,15 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
-import org.apache.ignite.internal.affinity.AffinityUtils;
import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
@@ -58,7 +51,6 @@ import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopolog
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
@@ -66,7 +58,6 @@ import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetastorageGroupId;
import org.apache.ignite.internal.placementdriver.leases.Lease;
-import org.apache.ignite.internal.placementdriver.leases.LeaseBatch;
import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessageGroup;
@@ -74,16 +65,11 @@ import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessage
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupServiceFactory;
import org.apache.ignite.internal.raft.configuration.RaftConfiguration;
-import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
-import org.apache.ignite.internal.schema.configuration.ExtendedTableChange;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
-import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.internal.vault.persistence.PersistentVaultService;
-import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
@@ -102,7 +88,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
* There are tests for Placement driver manager.
*/
@ExtendWith(ConfigurationExtension.class)
-public class PlacementDriverManagerTest extends IgniteAbstractTest {
+public class PlacementDriverManagerTest extends BasePlacementDriverTest {
public static final int PORT = 1234;
private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();
@@ -126,12 +112,6 @@ public class PlacementDriverManagerTest extends IgniteAbstractTest {
@InjectConfiguration
private RaftConfiguration raftConfiguration;
- @InjectConfiguration
- private TablesConfiguration tblsCfg;
-
- @InjectConfiguration("mock.distributionZones {zone1 = { partitions = 1, replicas = 2, zoneId = 1}}")
- private DistributionZonesConfiguration dstZnsCfg;
-
@InjectConfiguration
private MetaStorageConfiguration metaStorageConfiguration;
@@ -212,16 +192,15 @@ public class PlacementDriverManagerTest extends IgniteAbstractTest {
);
placementDriverManager = new PlacementDriverManager(
+ nodeName,
metaStorageManager,
vaultManager,
MetastorageGroupId.INSTANCE,
clusterService,
- () -> cmgManager.metaStorageNodes(),
+ cmgManager::metaStorageNodes,
logicalTopologyService,
raftManager,
topologyAwareRaftGroupServiceFactory,
- tblsCfg,
- dstZnsCfg,
clock
);
@@ -486,47 +465,9 @@ public class PlacementDriverManagerTest extends IgniteAbstractTest {
* Creates an assignment for the fake table.
*
* @return Replication group id.
- * @throws Exception If failed.
*/
- private TablePartitionId createTableAssignment() throws Exception {
- int tableId = nextTableId.incrementAndGet();
-
- List<Set<Assignment>> assignments = AffinityUtils.calculateAssignments(List.of(nodeName, anotherNodeName), 1, 2);
-
- tblsCfg.tables().change(tableViewTableChangeNamedListChange -> {
- tableViewTableChangeNamedListChange.create("test-table", tableChange -> {
- var extConfCh = ((ExtendedTableChange) tableChange);
-
- extConfCh.changeId(tableId);
-
- extConfCh.changeZoneId(getZoneId(dstZnsCfg, "zone1"));
- });
- }).thenCompose(v -> {
- Map<ByteArray, byte[]> partitionAssignments = new HashMap<>(assignments.size());
-
- for (int i = 0; i < assignments.size(); i++) {
- partitionAssignments.put(
- stablePartAssignmentsKey(
- new TablePartitionId(tableId, i)),
- ByteUtils.toBytes(assignments.get(i)));
-
- }
-
- return metaStorageManager.putAll(partitionAssignments);
- })
- .get();
-
- var grpPart0 = new TablePartitionId(tableId, 0);
-
- log.info("Fake table created [id={}, repGrp={}]", tableId, grpPart0);
-
- return grpPart0;
- }
-
- private Lease leaseFromBytes(byte[] bytes, ReplicationGroupId groupId) {
- LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN));
-
- return leaseBatch.leases().stream().filter(l -> l.replicationGroupId().equals(groupId)).findAny().orElse(null);
+ private TablePartitionId createTableAssignment() {
+ return createTableAssignment(metaStorageManager, nextTableId.incrementAndGet(), List.of(nodeName, anotherNodeName));
}
/**
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
index af1f8125bc..f99669bc89 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
@@ -35,7 +35,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.affinity.Assignment;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -44,17 +43,17 @@ import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.placementdriver.leases.Lease;
import org.apache.ignite.internal.placementdriver.leases.LeaseBatch;
import org.apache.ignite.internal.placementdriver.leases.LeaseTracker;
+import org.apache.ignite.internal.placementdriver.leases.Leases;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverActorMessage;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessageGroup;
import org.apache.ignite.internal.placementdriver.message.StopLeaseProlongationMessage;
import org.apache.ignite.internal.placementdriver.negotiation.LeaseAgreement;
import org.apache.ignite.internal.placementdriver.negotiation.LeaseNegotiator;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
-import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.thread.IgniteThread;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.vault.VaultManager;
-import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteSystemProperties;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
@@ -105,35 +104,34 @@ public class LeaseUpdater {
private final Updater updater;
/** Dedicated thread to update leases. */
- private volatile Thread updaterTread;
+ private volatile IgniteThread updaterThread;
/** Processor to communicate with the leaseholder to negotiate the lease. */
private LeaseNegotiator leaseNegotiator;
/** Node name. */
- private String nodeName;
+ private final String nodeName;
/**
- * The constructor.
+ * Constructor.
*
* @param clusterService Cluster service.
* @param vaultManager Vault manager.
* @param msManager Meta storage manager.
* @param topologyService Topology service.
- * @param tablesConfiguration Tables configuration.
* @param leaseTracker Lease tracker.
* @param clock Cluster clock.
*/
- public LeaseUpdater(
+ LeaseUpdater(
+ String nodeName,
ClusterService clusterService,
VaultManager vaultManager,
MetaStorageManager msManager,
LogicalTopologyService topologyService,
- TablesConfiguration tablesConfiguration,
- DistributionZonesConfiguration distributionZonesConfiguration,
LeaseTracker leaseTracker,
HybridClock clock
) {
+ this.nodeName = nodeName;
this.clusterService = clusterService;
this.msManager = msManager;
this.leaseTracker = leaseTracker;
@@ -147,28 +145,20 @@ public class LeaseUpdater {
clusterService.messagingService().addMessageHandler(PlacementDriverMessageGroup.class, new PlacementDriverActorMessageHandler());
}
- /**
- * Initializes the class.
- */
- public void init(String nodeName) {
- this.nodeName = nodeName;
-
+ /** Initializes the class. */
+ public void init() {
topologyTracker.startTrack();
assignmentsTracker.startTrack();
}
- /**
- * De-initializes the class.
- */
- public void deInit() {
+ /** De-initializes the class. */
+ void deInit() {
topologyTracker.stopTrack();
assignmentsTracker.stopTrack();
}
- /**
- * Activates a lease updater to renew leases.
- */
- public void activate() {
+ /** Activates a lease updater to renew leases. */
+ void activate() {
if (active()) {
return;
}
@@ -182,18 +172,16 @@ public class LeaseUpdater {
leaseNegotiator = new LeaseNegotiator(clusterService);
- updaterTread = new Thread(updater, NamedThreadFactory.threadPrefix(nodeName, "lease-updater") + '0');
+ updaterThread = new IgniteThread(nodeName, "lease-updater", updater);
- updaterTread.start();
+ updaterThread.start();
} finally {
stateChangingLock.unblock();
}
}
- /**
- * Stops a dedicated thread to renew or assign leases.
- */
- public void deactivate() {
+ /** Stops a dedicated thread to renew or assign leases. */
+ void deactivate() {
if (!active()) {
return;
}
@@ -207,10 +195,12 @@ public class LeaseUpdater {
leaseNegotiator = null;
+ Thread updaterTread = this.updaterThread;
+
if (updaterTread != null) {
updaterTread.interrupt();
- updaterTread = null;
+ this.updaterThread = null;
}
} finally {
@@ -225,15 +215,17 @@ public class LeaseUpdater {
* @param lease Lease to deny.
* @return Future completes true when the lease will not prolong in the future, false otherwise.
*/
- public CompletableFuture<Boolean> denyLease(ReplicationGroupId grpId, Lease lease) {
+ private CompletableFuture<Boolean> denyLease(ReplicationGroupId grpId, Lease lease) {
Lease deniedLease = lease.denyLease();
leaseNegotiator.onLeaseRemoved(grpId);
- IgniteBiTuple<Map<ReplicationGroupId, Lease>, byte[]> leasesCurrent = leaseTracker.leasesCurrent();
+ Leases leasesCurrent = leaseTracker.leasesCurrent();
+
+ Collection<Lease> leases = leasesCurrent.leaseByGroupId().values();
- Collection<Lease> leases = leasesCurrent.getKey().values();
List<Lease> renewedLeases = new ArrayList<>();
+
for (Lease ls : leases) {
if (ls.replicationGroupId().equals(grpId)) {
renewedLeases.add(deniedLease);
@@ -242,10 +234,10 @@ public class LeaseUpdater {
}
}
- var key = PLACEMENTDRIVER_LEASES_KEY;
+ ByteArray key = PLACEMENTDRIVER_LEASES_KEY;
return msManager.invoke(
- or(notExists(key), value(key).eq(leasesCurrent.getValue())),
+ or(notExists(key), value(key).eq(leasesCurrent.leasesBytes())),
put(key, new LeaseBatch(renewedLeases).bytes()),
noop()
);
@@ -258,7 +250,7 @@ public class LeaseUpdater {
* @param proposedConsistentId Proposed consistent id, found out of a lease negotiation. The parameter might be {@code null}.
* @return Cluster node, or {@code null} if no node in assignments can be the leaseholder.
*/
- private ClusterNode nextLeaseHolder(Set<Assignment> assignments, @Nullable String proposedConsistentId) {
+ private @Nullable ClusterNode nextLeaseHolder(Set<Assignment> assignments, @Nullable String proposedConsistentId) {
//TODO: IGNITE-18879 Implement more intellectual algorithm to choose a node.
String consistentId = null;
@@ -272,33 +264,19 @@ public class LeaseUpdater {
}
}
- if (consistentId != null) {
- ClusterNode candidate = topologyTracker.nodeByConsistentId(consistentId);
-
- if (candidate != null) {
- return candidate;
- }
- }
-
- return null;
+ return consistentId == null ? null : topologyTracker.nodeByConsistentId(consistentId);
}
- /**
- * Whether is active.
- *
- * @return Whether is active.
- */
- public boolean active() {
+ /** Returns {@code true} if active. */
+ boolean active() {
return active.get();
}
- /**
- * Runnable to update lease in Meta storage.
- */
+ /** Runnable to update lease in Meta storage. */
private class Updater implements Runnable {
@Override
public void run() {
- while (updaterTread != null && !updaterTread.isInterrupted()) {
+ while (updaterThread != null && !updaterThread.isInterrupted()) {
if (!active() || !stateChangingLock.enterBusy()) {
return;
}
@@ -317,15 +295,13 @@ public class LeaseUpdater {
}
}
- /**
- * Updates leases in Meta storage. This method is supposed to be used in the busy lock.
- */
+ /** Updates leases in Meta storage. This method is supposed to be used in the busy lock. */
private void updateLeaseBatchInternal() {
long outdatedLeaseThreshold = clock.now().getPhysical() + LEASE_INTERVAL / 2;
- IgniteBiTuple<Map<ReplicationGroupId, Lease>, byte[]> leasesCurrent = leaseTracker.leasesCurrent();
+ Leases leasesCurrent = leaseTracker.leasesCurrent();
Map<ReplicationGroupId, Boolean> toBeNegotiated = new HashMap<>();
- Map<ReplicationGroupId, Lease> renewedLeases = new HashMap<>(leasesCurrent.getKey());
+ Map<ReplicationGroupId, Lease> renewedLeases = new HashMap<>(leasesCurrent.leaseByGroupId());
for (Map.Entry<ReplicationGroupId, Set<Assignment>> entry : assignmentsTracker.assignments().entrySet()) {
ReplicationGroupId grpId = entry.getKey();
@@ -382,7 +358,7 @@ public class LeaseUpdater {
var key = PLACEMENTDRIVER_LEASES_KEY;
msManager.invoke(
- or(notExists(key), value(key).eq(leasesCurrent.getValue())),
+ or(notExists(key), value(key).eq(leasesCurrent.leasesBytes())),
put(key, renewedValue),
noop()
).thenAccept(success -> {
@@ -460,9 +436,7 @@ public class LeaseUpdater {
}
}
- /**
- * Message handler to process notification from replica side.
- */
+ /** Message handler to process notification from replica side. */
private class PlacementDriverActorMessageHandler implements NetworkMessageHandler {
@Override
public void onReceived(NetworkMessage msg0, String sender, @Nullable Long correlationId) {
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
index 277b0dab82..9c0b01e69d 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.placementdriver;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -27,7 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
-import org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -39,11 +38,9 @@ import org.apache.ignite.internal.raft.RaftManager;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupServiceFactory;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
-import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
@@ -55,7 +52,9 @@ import org.jetbrains.annotations.TestOnly;
* The another role of the manager is providing a node, which is leaseholder at the moment, for a particular replication group.
*/
public class PlacementDriverManager implements IgniteComponent {
- public static final String PLACEMENTDRIVER_LEASES_KEY_STRING = "placementdriver.leases";
+ private static final IgniteLogger LOG = Loggers.forClass(PlacementDriverManager.class);
+
+ private static final String PLACEMENTDRIVER_LEASES_KEY_STRING = "placementdriver.leases";
public static final ByteArray PLACEMENTDRIVER_LEASES_KEY = ByteArray.fromString(PLACEMENTDRIVER_LEASES_KEY_STRING);
@@ -63,7 +62,7 @@ public class PlacementDriverManager implements IgniteComponent {
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
/** Prevents double stopping of the component. */
- private final AtomicBoolean isStopped = new AtomicBoolean();
+ private final AtomicBoolean stopGuard = new AtomicBoolean();
/** Cluster service. */
private final ClusterService clusterService;
@@ -90,8 +89,9 @@ public class PlacementDriverManager implements IgniteComponent {
private final LeaseUpdater leaseUpdater;
/**
- * The constructor.
+ * Constructor.
*
+ * @param nodeName Node name.
* @param metaStorageMgr Meta Storage manager.
* @param vaultManager Vault manager.
* @param replicationGroupId Id of placement driver group.
@@ -100,10 +100,10 @@ public class PlacementDriverManager implements IgniteComponent {
* @param logicalTopologyService Logical topology service.
* @param raftManager Raft manager.
* @param topologyAwareRaftGroupServiceFactory Raft client factory.
- * @param tablesCfg Table configuration.
* @param clock Hybrid clock.
*/
public PlacementDriverManager(
+ String nodeName,
MetaStorageManager metaStorageMgr,
VaultManager vaultManager,
ReplicationGroupId replicationGroupId,
@@ -112,8 +112,6 @@ public class PlacementDriverManager implements IgniteComponent {
LogicalTopologyService logicalTopologyService,
RaftManager raftManager,
TopologyAwareRaftGroupServiceFactory topologyAwareRaftGroupServiceFactory,
- TablesConfiguration tablesCfg,
- DistributionZonesConfiguration distributionZonesConfiguration,
HybridClock clock
) {
this.replicationGroupId = replicationGroupId;
@@ -126,58 +124,54 @@ public class PlacementDriverManager implements IgniteComponent {
this.leaseTracker = new LeaseTracker(vaultManager, metaStorageMgr);
this.leaseUpdater = new LeaseUpdater(
+ nodeName,
clusterService,
vaultManager,
metaStorageMgr,
logicalTopologyService,
- tablesCfg,
- distributionZonesConfiguration,
leaseTracker,
clock
);
}
- /** {@inheritDoc} */
@Override
public void start() {
- placementDriverNodesNamesProvider.get()
- .thenCompose(placementDriverNodes -> {
- String thisNodeName = clusterService.topologyService().localMember().name();
+ inBusyLock(busyLock, () -> {
+ placementDriverNodesNamesProvider.get()
+ .thenCompose(placementDriverNodes -> {
+ String thisNodeName = clusterService.topologyService().localMember().name();
+
+ if (!placementDriverNodes.contains(thisNodeName)) {
+ return completedFuture(null);
+ }
- if (placementDriverNodes.contains(thisNodeName)) {
try {
- leaseUpdater.init(thisNodeName);
+ leaseUpdater.init();
return raftManager.startRaftGroupService(
replicationGroupId,
PeersAndLearners.fromConsistentIds(placementDriverNodes),
topologyAwareRaftGroupServiceFactory
- ).thenCompose(client -> client.subscribeLeader(this::onLeaderChange).thenApply(v -> client));
+ ).thenCompose(client -> client.subscribeLeader(this::onLeaderChange).thenApply(v -> client));
} catch (NodeStoppingException e) {
return failedFuture(e);
}
- } else {
- return completedFuture(null);
- }
- })
- .whenComplete((client, ex) -> {
- if (ex == null) {
- raftClientFuture.complete(client);
- } else {
- raftClientFuture.completeExceptionally(ex);
- }
- });
-
- leaseTracker.startTrack();
+ })
+ .whenComplete((client, ex) -> {
+ if (ex == null) {
+ raftClientFuture.complete(client);
+ } else {
+ raftClientFuture.completeExceptionally(ex);
+ }
+ });
+
+ leaseTracker.startTrack();
+ });
}
- /** {@inheritDoc} */
@Override
public void beforeNodeStop() {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
- try {
+ inBusyLock(busyLock, () -> {
withRaftClientIfPresent(c -> {
c.unsubscribeLeader().join();
@@ -185,15 +179,12 @@ public class PlacementDriverManager implements IgniteComponent {
});
leaseTracker.stopTrack();
- } finally {
- busyLock.leaveBusy();
- }
+ });
}
- /** {@inheritDoc} */
@Override
public void stop() throws Exception {
- if (!isStopped.compareAndSet(false, true)) {
+ if (!stopGuard.compareAndSet(false, true)) {
return;
}
@@ -213,37 +204,24 @@ public class PlacementDriverManager implements IgniteComponent {
}
private void onLeaderChange(ClusterNode leader, long term) {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
- try {
+ inBusyLock(busyLock, () -> {
if (leader.equals(clusterService.topologyService().localMember())) {
- takeOverActiveActor();
+ takeOverActiveActorBusy();
} else {
- stepDownActiveActor();
+ stepDownActiveActorBusy();
}
- } finally {
- busyLock.leaveBusy();
- }
+ });
}
- /** The logger. */
- private static IgniteLogger LOG = Loggers.forClass(PlacementDriverManager.class);
-
- /**
- * Takes over active actor of placement driver group.
- */
- private void takeOverActiveActor() {
+ /** Takes over active actor of placement driver group. */
+ private void takeOverActiveActorBusy() {
LOG.info("Placement driver active actor is starting.");
leaseUpdater.activate();
}
-
- /**
- * Steps down as active actor.
- */
- private void stepDownActiveActor() {
+ /** Steps down as active actor. */
+ private void stepDownActiveActorBusy() {
LOG.info("Placement driver active actor is stopping.");
leaseUpdater.deactivate();
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/TopologyTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/TopologyTracker.java
index 293126c3af..ba9e5ffbdf 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/TopologyTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/TopologyTracker.java
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.util.CollectionUtils;
import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
/**
* The class tracks a logical topology.
@@ -89,7 +90,7 @@ public class TopologyTracker {
* @param consistentId Node consistent id.
* @return Cluster node or {@code null} if topology has no a node with the consistent id.
*/
- public ClusterNode nodeByConsistentId(String consistentId) {
+ public @Nullable ClusterNode nodeByConsistentId(String consistentId) {
LogicalTopologySnapshot logicalTopologySnap0 = topologySnapRef.get();
if (logicalTopologySnap0 == null || CollectionUtils.nullOrEmpty(logicalTopologySnap0.nodes())) {
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
index a3a707c84a..22b1e0e8c2 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
@@ -17,17 +17,18 @@
package org.apache.ignite.internal.placementdriver.leases;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
import static java.util.Collections.emptyMap;
import static java.util.Collections.unmodifiableMap;
import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.hlc.HybridTimestamp.MIN_VALUE;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_LEASES_KEY;
import static org.apache.ignite.internal.placementdriver.leases.Lease.EMPTY_LEASE;
+import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -51,9 +52,6 @@ import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.PendingIndependentComparableValuesTracker;
import org.apache.ignite.internal.vault.VaultEntry;
import org.apache.ignite.internal.vault.VaultManager;
-import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.lang.NodeStoppingException;
-import org.jetbrains.annotations.NotNull;
/**
* Class tracks cluster leases in memory.
@@ -73,19 +71,20 @@ public class LeaseTracker implements PlacementDriver {
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
/** Prevents double stopping of the tracker. */
- private final AtomicBoolean isStopped = new AtomicBoolean();
+ private final AtomicBoolean stopGuard = new AtomicBoolean();
/** Leases cache. */
- private volatile IgniteBiTuple<Map<ReplicationGroupId, Lease>, byte[]> leases;
+ private volatile Leases leases = new Leases(emptyMap(), BYTE_EMPTY_ARRAY);
/** Map of primary replica waiters. */
- private final Map<ReplicationGroupId, PendingIndependentComparableValuesTracker<HybridTimestamp, LeaseMeta>> primaryReplicaWaiters;
+ private final Map<ReplicationGroupId, PendingIndependentComparableValuesTracker<HybridTimestamp, LeaseMeta>> primaryReplicaWaiters
+ = new ConcurrentHashMap<>();
/** Listener to update a leases cache. */
private final UpdateListener updateListener = new UpdateListener();
/**
- * The constructor.
+ * Constructor.
*
* @param vaultManager Vault manager.
* @param msManager Meta storage manager.
@@ -93,50 +92,44 @@ public class LeaseTracker implements PlacementDriver {
public LeaseTracker(VaultManager vaultManager, MetaStorageManager msManager) {
this.vaultManager = vaultManager;
this.msManager = msManager;
-
- this.leases = new IgniteBiTuple<>(emptyMap(), new byte[0]);
- this.primaryReplicaWaiters = new ConcurrentHashMap<>();
}
- /**
- * Recoveries state from Vault and subscribers on further updates.
- */
+ /** Recoveries state from Vault and subscribers on further updates. */
public void startTrack() {
- msManager.registerPrefixWatch(PLACEMENTDRIVER_LEASES_KEY, updateListener);
+ inBusyLock(busyLock, () -> {
+ msManager.registerPrefixWatch(PLACEMENTDRIVER_LEASES_KEY, updateListener);
- CompletableFuture<VaultEntry> entryFut = vaultManager.get(PLACEMENTDRIVER_LEASES_KEY);
+ CompletableFuture<VaultEntry> entryFut = vaultManager.get(PLACEMENTDRIVER_LEASES_KEY);
- VaultEntry entry = entryFut.join();
+ VaultEntry entry = entryFut.join();
- Map<ReplicationGroupId, Lease> leasesMap = new HashMap<>();
+ Map<ReplicationGroupId, Lease> leasesMap = new HashMap<>();
- byte[] leasesBytes;
+ byte[] leasesBytes;
- if (entry != null) {
- leasesBytes = entry.value();
+ if (entry != null) {
+ leasesBytes = entry.value();
- LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(leasesBytes).order(ByteOrder.LITTLE_ENDIAN));
+ LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(leasesBytes).order(LITTLE_ENDIAN));
- leaseBatch.leases().forEach(lease -> {
- leasesMap.put(lease.replicationGroupId(), lease);
- primaryReplicaWaiters.computeIfAbsent(lease.replicationGroupId(),
- groupId -> new PendingIndependentComparableValuesTracker<>(MIN_VALUE))
- .update(lease.getExpirationTime(), lease);
- });
- } else {
- leasesBytes = new byte[0];
- }
+ leaseBatch.leases().forEach(lease -> {
+ leasesMap.put(lease.replicationGroupId(), lease);
+
+ getOrCreatePrimaryReplicaWaiter(lease.replicationGroupId()).update(lease.getExpirationTime(), lease);
+ });
+ } else {
+ leasesBytes = BYTE_EMPTY_ARRAY;
+ }
- leases = new IgniteBiTuple<>(unmodifiableMap(leasesMap), leasesBytes);
+ leases = new Leases(unmodifiableMap(leasesMap), leasesBytes);
- LOG.info("Leases cache recovered [leases={}]", leases);
+ LOG.info("Leases cache recovered [leases={}]", leases);
+ });
}
- /**
- * Stops the tracker.
- */
+ /** Stops the tracker. */
public void stopTrack() {
- if (!isStopped.compareAndSet(false, true)) {
+ if (!stopGuard.compareAndSet(false, true)) {
return;
}
@@ -154,60 +147,59 @@ public class LeaseTracker implements PlacementDriver {
* @param grpId Replication group id.
* @return A lease is associated with the group.
*/
- public @NotNull Lease getLease(ReplicationGroupId grpId) {
+ public Lease getLease(ReplicationGroupId grpId) {
+ Leases leases = this.leases;
+
assert leases != null : "Leases not initialized, probably the local placement driver actor hasn't started lease tracking.";
- return leases.get1().getOrDefault(grpId, EMPTY_LEASE);
+ return leases.leaseByGroupId().getOrDefault(grpId, EMPTY_LEASE);
}
- /**
- * Returns collection of leases, ordered by replication group.
- *
- * @return Collection of leases.
- */
- public IgniteBiTuple<Map<ReplicationGroupId, Lease>, byte[]> leasesCurrent() {
+ /** Returns collection of leases, ordered by replication group. */
+ public Leases leasesCurrent() {
return leases;
}
- /**
- * Listen lease holder updates.
- */
+ /** Listen lease holder updates. */
private class UpdateListener implements WatchListener {
@Override
public CompletableFuture<Void> onUpdate(WatchEvent event) {
- for (EntryEvent entry : event.entryEvents()) {
- Entry msEntry = entry.newEntry();
+ return inBusyLockAsync(busyLock, () -> {
+ for (EntryEvent entry : event.entryEvents()) {
+ Entry msEntry = entry.newEntry();
- byte[] leasesBytes = msEntry.value();
- Map<ReplicationGroupId, Lease> leasesMap = new HashMap<>();
+ byte[] leasesBytes = msEntry.value();
+ Map<ReplicationGroupId, Lease> leasesMap = new HashMap<>();
- LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(leasesBytes).order(ByteOrder.LITTLE_ENDIAN));
+ LeaseBatch leaseBatch = LeaseBatch.fromBytes(ByteBuffer.wrap(leasesBytes).order(LITTLE_ENDIAN));
- Set<ReplicationGroupId> actualGroups = new HashSet<>();
+ Set<ReplicationGroupId> actualGroups = new HashSet<>();
- for (Lease lease : leaseBatch.leases()) {
- ReplicationGroupId grpId = lease.replicationGroupId();
- actualGroups.add(grpId);
+ for (Lease lease : leaseBatch.leases()) {
+ ReplicationGroupId grpId = lease.replicationGroupId();
+ actualGroups.add(grpId);
- leasesMap.put(grpId, lease);
+ leasesMap.put(grpId, lease);
- primaryReplicaWaiters.computeIfAbsent(grpId, groupId -> new PendingIndependentComparableValuesTracker<>(MIN_VALUE))
- .update(lease.getExpirationTime(), lease);
- }
+ primaryReplicaWaiters
+ .computeIfAbsent(grpId, groupId -> new PendingIndependentComparableValuesTracker<>(MIN_VALUE))
+ .update(lease.getExpirationTime(), lease);
+ }
- for (Iterator<Map.Entry<ReplicationGroupId, Lease>> iterator = leasesMap.entrySet().iterator(); iterator.hasNext();) {
- Map.Entry<ReplicationGroupId, Lease> e = iterator.next();
+ for (Iterator<Map.Entry<ReplicationGroupId, Lease>> iterator = leasesMap.entrySet().iterator(); iterator.hasNext();) {
+ Map.Entry<ReplicationGroupId, Lease> e = iterator.next();
- if (!actualGroups.contains(e.getKey())) {
- iterator.remove();
- tryRemoveTracker(e.getKey());
+ if (!actualGroups.contains(e.getKey())) {
+ iterator.remove();
+ tryRemoveTracker(e.getKey());
+ }
}
- }
- LeaseTracker.this.leases = new IgniteBiTuple<>(unmodifiableMap(leasesMap), leasesBytes);
- }
+ LeaseTracker.this.leases = new Leases(unmodifiableMap(leasesMap), leasesBytes);
+ }
- return completedFuture(null);
+ return completedFuture(null);
+ });
}
@Override
@@ -217,34 +209,24 @@ public class LeaseTracker implements PlacementDriver {
@Override
public CompletableFuture<LeaseMeta> awaitPrimaryReplica(ReplicationGroupId groupId, HybridTimestamp timestamp) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new NodeStoppingException("Component is stopping."));
- }
- try {
- return primaryReplicaWaiters.computeIfAbsent(groupId, id -> new PendingIndependentComparableValuesTracker<>(MIN_VALUE))
- .waitFor(timestamp);
- } finally {
- busyLock.leaveBusy();
- }
+ return inBusyLockAsync(busyLock, () -> getOrCreatePrimaryReplicaWaiter(groupId).waitFor(timestamp));
}
@Override
public CompletableFuture<LeaseMeta> getPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
- if (!busyLock.enterBusy()) {
- return failedFuture(new NodeStoppingException("Component is stopping."));
- }
+ return inBusyLockAsync(busyLock, () -> {
+ Map<ReplicationGroupId, Lease> leasesMap = leases.leaseByGroupId();
- Map<ReplicationGroupId, Lease> leasesMap = leases.get1();
-
- try {
Lease lease = leasesMap.getOrDefault(replicationGroupId, EMPTY_LEASE);
if (lease.getExpirationTime().after(timestamp)) {
return completedFuture(lease);
}
- return msManager.clusterTime().waitFor(timestamp).thenApply(ignored -> inBusyLock(
- busyLock, () -> {
+ return msManager
+ .clusterTime()
+ .waitFor(timestamp)
+ .thenApply(ignored -> inBusyLock(busyLock, () -> {
Lease lease0 = leasesMap.getOrDefault(replicationGroupId, EMPTY_LEASE);
if (lease0.getExpirationTime().after(timestamp)) {
@@ -253,9 +235,7 @@ public class LeaseTracker implements PlacementDriver {
return null;
}
}));
- } finally {
- busyLock.leaveBusy();
- }
+ });
}
/**
@@ -273,4 +253,10 @@ public class LeaseTracker implements PlacementDriver {
return tracker0;
});
}
+
+ private PendingIndependentComparableValuesTracker<HybridTimestamp, LeaseMeta> getOrCreatePrimaryReplicaWaiter(
+ ReplicationGroupId groupId
+ ) {
+ return primaryReplicaWaiters.computeIfAbsent(groupId, key -> new PendingIndependentComparableValuesTracker<>(MIN_VALUE));
+ }
}
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/Leases.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/Leases.java
new file mode 100644
index 0000000000..9a0e6ca6ad
--- /dev/null
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/Leases.java
@@ -0,0 +1,43 @@
+/*
+ * 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.placementdriver.leases;
+
+import java.util.Map;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+
+/** Leases received from the metastore. */
+public class Leases {
+ private final Map<ReplicationGroupId, Lease> leaseByGroupId;
+
+ private final byte[] leasesBytes;
+
+ Leases(Map<ReplicationGroupId, Lease> leaseByGroupId, byte[] leasesBytes) {
+ this.leaseByGroupId = leaseByGroupId;
+ this.leasesBytes = leasesBytes;
+ }
+
+ /** Returns leases grouped by replication group. */
+ public Map<ReplicationGroupId, Lease> leaseByGroupId() {
+ return leaseByGroupId;
+ }
+
+ /** Returns an array of byte leases from the metastore. */
+ public byte[] leasesBytes() {
+ return leasesBytes;
+ }
+}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 314b22f4d8..ab18f6a3ea 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -451,6 +451,7 @@ public class IgniteImpl implements Ignite {
metaStorageMgr.configure(clusterConfigRegistry.getConfiguration(MetaStorageConfiguration.KEY));
placementDriverMgr = new PlacementDriverManager(
+ name,
metaStorageMgr,
vaultMgr,
MetastorageGroupId.INSTANCE,
@@ -459,8 +460,6 @@ public class IgniteImpl implements Ignite {
logicalTopologyService,
raftMgr,
topologyAwareRaftGroupServiceFactory,
- tablesConfig,
- zonesConfig,
clock
);