You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "kgusakov (via GitHub)" <gi...@apache.org> on 2024/06/02 23:12:57 UTC
[PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
kgusakov opened a new pull request, #3858:
URL: https://github.com/apache/ignite-3/pull/3858
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644759672
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
Review Comment:
`Not only in case of NodeStoppingException.` what do you mean? catch all exceptions, why? What about the failed future - done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "alievmirza (via GitHub)" <gi...@apache.org>.
alievmirza commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1625774312
##########
modules/replica-lifecycle/src/integrationTest/java/org/apache/ignite/internal/datareplication/utils/ZoneBasedPlacementDriver.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.datareplication.utils;
+
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.event.AbstractEventProducer;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Trivial placement driver for tests.
+ */
+public class ZoneBasedPlacementDriver extends AbstractEventProducer<PrimaryReplicaEvent, PrimaryReplicaEventParameters> implements
Review Comment:
redundant whitespace
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ private final Marshaller raftCommandsMarshaller;
+
+ public static final String ZONE_STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(ReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param raftMgr RAFT manager.
+ * @param replicaMgr Replica manager.
+ * @param raftGroupServiceFactory Raft clients factory.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ * @param messageSerializationRegistry Message serialization registry.
+ */
+ public ReplicaLifecycleManager(CatalogManager catalogMgr, RaftManager raftMgr, ReplicaManager replicaMgr,
+ TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory, DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr, TopologyService topologyService,
+ MessageSerializationRegistry messageSerializationRegistry) {
+ this.catalogMgr = catalogMgr;
+ this.raftMgr = raftMgr;
+ this.replicaMgr = replicaMgr;
+ this.raftGroupServiceFactory = raftGroupServiceFactory;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ this.raftCommandsMarshaller = new ThreadLocalPartitionCommandsMarshaller(messageSerializationRegistry);
+ }
@Override
public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
if (!ENABLED) {
return nullCompletedFuture();
}
- // TODO: IGNITE-22231 Will be replaced by the real code.
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
return nullCompletedFuture();
}
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ Peer serverPeer = realConfiguration.peer(localNode().name());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-22391 support for volatile stores
+ RaftGroupOptions raftGroupOptions = RaftGroupOptions.forPersistentStores();
+
+ raftGroupOptions.snapshotStorageFactory(new FailFastSnapshotStorageFactory());
+
+ raftGroupOptions.commandsMarshaller(raftCommandsMarshaller);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ CompletableFuture<TopologyAwareRaftGroupService> raftClient = ((Loza) raftMgr).startRaftGroupNode(
Review Comment:
redundant whitespace
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/PartitionGroupListener.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.datareplication;
+
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.raft.ReadCommand;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.service.CommandClosure;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+
+/**
+ * Listener for the zone partitoin.
Review Comment:
typo
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/snapshot/FailFastSnapshotStorageFactory.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.datareplication.snapshot;
+
+import org.apache.ignite.internal.raft.storage.SnapshotStorageFactory;
+import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.option.SnapshotCopierOptions;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
+import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotCopier;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
+
+/**
+ * Temporary snapshot factory, which must help to identify the unexpected snapshots,
+ * while zone based replicas is not support snapshotting yet.
+ */
+// TODO https://issues.apache.org/jira/browse/IGNITE-22115 remove it
Review Comment:
I would expect here link not to the epic, but to a specific ticket
##########
modules/replicator/src/main/java/org/apache/ignite/internal/ZonePartitionReplica.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
+import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
+import org.apache.ignite.internal.replicator.Replica;
+import org.apache.ignite.internal.replicator.ReplicaResult;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.jetbrains.annotations.TestOnly;
+
+/**
+ * Replica for the zone based partitions.
+ */
+public class ZonePartitionReplica extends Replica {
Review Comment:
Why do we need this? Will it be removed after all work for single replica is done? If yes, I would expect deprecated annotation here
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -154,6 +156,47 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
private String localNodeId;
+ private Function<ReplicationGroupId, ReplicationGroupId> groupIdConverter = Function.identity();
Review Comment:
java doc is required, we need to know how conversion works
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ private final Marshaller raftCommandsMarshaller;
+
+ public static final String ZONE_STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(ReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param raftMgr RAFT manager.
+ * @param replicaMgr Replica manager.
+ * @param raftGroupServiceFactory Raft clients factory.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ * @param messageSerializationRegistry Message serialization registry.
+ */
+ public ReplicaLifecycleManager(CatalogManager catalogMgr, RaftManager raftMgr, ReplicaManager replicaMgr,
+ TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory, DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr, TopologyService topologyService,
+ MessageSerializationRegistry messageSerializationRegistry) {
+ this.catalogMgr = catalogMgr;
+ this.raftMgr = raftMgr;
+ this.replicaMgr = replicaMgr;
+ this.raftGroupServiceFactory = raftGroupServiceFactory;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ this.raftCommandsMarshaller = new ThreadLocalPartitionCommandsMarshaller(messageSerializationRegistry);
+ }
@Override
public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
if (!ENABLED) {
return nullCompletedFuture();
}
- // TODO: IGNITE-22231 Will be replaced by the real code.
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
return nullCompletedFuture();
}
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ Peer serverPeer = realConfiguration.peer(localNode().name());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ var raftNodeId = new RaftNodeId(replicaGrpId, serverPeer);
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-22391 support for volatile stores
+ RaftGroupOptions raftGroupOptions = RaftGroupOptions.forPersistentStores();
+
+ raftGroupOptions.snapshotStorageFactory(new FailFastSnapshotStorageFactory());
+
+ raftGroupOptions.commandsMarshaller(raftCommandsMarshaller);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ CompletableFuture<TopologyAwareRaftGroupService> raftClient = ((Loza) raftMgr).startRaftGroupNode(
+ raftNodeId,
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr,
+ raftGroupOptions,
+ raftGroupServiceFactory
+ );
+
+ return raftClient.thenCompose(client ->
+ replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ client
+ )).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
+ var peers = new HashSet<String>();
+ var learners = new HashSet<String>();
+
+ for (Assignment assignment : assignments) {
+ if (assignment.isPeer()) {
+ peers.add(assignment.consistentId());
+ } else {
+ learners.add(assignment.consistentId());
+ }
+ }
+
+ return PeersAndLearners.fromConsistentIds(peers, learners);
+ }
+
+ private ClusterNode localNode() {
+ return topologyService.localMember();
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ for (ReplicationGroupId replicationGroupId : replicationGroupIds) {
+ try {
+ raftMgr.stopRaftNodes(replicationGroupId);
+ replicaMgr.stopReplica(replicationGroupId);
+ } catch (NodeStoppingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * 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 zoneId Zone id.
+ * @param assignmentsFuture Assignments future, to get the assignments that should be written.
+ * @return Real list of assignments.
+ */
+ private CompletableFuture<List<Assignments>> writeTableAssignmentsToMetastore(
+ int zoneId,
+ CompletableFuture<List<Assignments>> assignmentsFuture
+ ) {
+ return assignmentsFuture.thenCompose(newAssignments -> {
+ assert !newAssignments.isEmpty();
+
+ List<Operation> partitionAssignments = new ArrayList<>(newAssignments.size());
+
+ for (int i = 0; i < newAssignments.size(); i++) {
+ ByteArray stableAssignmentsKey = stablePartAssignmentsKey(new ZonePartitionId(zoneId, i));
+ byte[] anAssignment = newAssignments.get(i).toBytes();
+ Operation op = put(stableAssignmentsKey, anAssignment);
+ partitionAssignments.add(op);
+ }
+
+ Condition condition = notExists(new ByteArray(partitionAssignments.get(0).key()));
+
+ return metaStorageMgr
+ .invoke(condition, partitionAssignments, Collections.emptyList())
+ .handle((invokeResult, e) -> {
+ if (e != null) {
+ LOG.error(
+ "Couldn't write assignments [assignmentsList={}] to metastore during invoke.",
+ e,
+ Assignments.assignmentListToString(newAssignments)
+ );
+
+ throw ExceptionUtils.sneakyThrow(e);
+ }
+
+ return invokeResult;
+ })
+ .thenCompose(invokeResult -> {
+ if (invokeResult) {
+ LOG.info(
+ "Assignments calculated from data nodes are successfully written to meta storage"
+ + " [zoneId={}, assignments={}].",
+ zoneId,
+ Assignments.assignmentListToString(newAssignments)
+ );
+
+ return completedFuture(newAssignments);
+ } else {
+ Set<ByteArray> partKeys = IntStream.range(0, newAssignments.size())
+ .mapToObj(p -> stablePartAssignmentsKey(new ZonePartitionId(zoneId, p)))
+ .collect(toSet());
+
+ CompletableFuture<Map<ByteArray, Entry>> resFuture = metaStorageMgr.getAll(partKeys);
+
+ return resFuture.thenApply(metaStorageAssignments -> {
+ List<Assignments> realAssignments = new ArrayList<>();
+
+ for (int p = 0; p < newAssignments.size(); p++) {
+ var partId = new ZonePartitionId(zoneId, p);
+ Entry assignmentsEntry = metaStorageAssignments.get(stablePartAssignmentsKey(partId));
+
+ assert assignmentsEntry != null && !assignmentsEntry.empty() && !assignmentsEntry.tombstone()
+ : "Unexpected assignments for partition [" + partId + ", entry=" + assignmentsEntry + "].";
+
+ Assignments real = Assignments.fromBytes(assignmentsEntry.value());
+
+ realAssignments.add(real);
+ }
+
+ LOG.info(
+ "Assignments picked up from meta storage [zoneId={}, assignments={}].",
+ zoneId,
+ Assignments.assignmentListToString(realAssignments)
+ );
+
+ return realAssignments;
+ });
+ }
+ })
+ .handle((realAssignments, e) -> {
+ if (e != null) {
+ LOG.error("Couldn't get assignments from metastore for zone [zoneId={}].", e, zoneId);
+
+ throw ExceptionUtils.sneakyThrow(e);
+ }
+
+ return realAssignments;
+ });
+ });
+ }
+
+ /**
+ * Check if the table already has assignments in the meta storage locally.
Review Comment:
Lets write correct javadocs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644610094
##########
modules/partition-replica/build.gradle:
##########
@@ -0,0 +1,78 @@
+/*
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1628416066
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -154,6 +156,47 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
private String localNodeId;
+ private Function<ReplicationGroupId, ReplicationGroupId> groupIdConverter = Function.identity();
Review Comment:
How it works - it's the responsibility of the call side. I added the same javadoc as already have constructor param, to explain what's the purpose of the field here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644406403
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java:
##########
@@ -17,126 +17,22 @@
package org.apache.ignite.internal.replicator;
-import static java.lang.System.currentTimeMillis;
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
-import static org.apache.ignite.internal.util.IgniteUtils.retryOperationUntilSuccess;
-
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiFunction;
-import org.apache.ignite.internal.hlc.ClockService;
-import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.lang.IgniteStringFormatter;
-import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.placementdriver.PlacementDriver;
-import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
-import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
-import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
-import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
-import org.apache.ignite.internal.replicator.listener.ReplicaListener;
-import org.apache.ignite.internal.replicator.message.PrimaryReplicaChangeCommand;
-import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaRequest;
-import org.apache.ignite.internal.util.PendingComparableValuesTracker;
-import org.apache.ignite.network.ClusterNode;
/**
- * Replica server.
+ * Interface for replica server.
*/
-public class Replica {
- /** The logger. */
- private static final IgniteLogger LOG = Loggers.forClass(ReplicaManager.class);
-
- /** Message factory. */
- private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();
-
- private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
-
- /** Replica group identity, this id is the same as the considered partition's id. */
- private final ReplicationGroupId replicaGrpId;
-
- /** Replica listener. */
- private final ReplicaListener listener;
-
- /** Storage index tracker. */
- private final PendingComparableValuesTracker<Long, Void> storageIndexTracker;
-
- /** Topology aware Raft client. */
- private final TopologyAwareRaftGroupService raftClient;
-
- /** Instance of the local node. */
- private final ClusterNode localNode;
-
- // TODO IGNITE-19120 after replica inoperability logic is introduced, this future should be replaced with something like
- // VersionedValue (so that PlacementDriverMessages would wait for new leader election)
- /** Completes when leader is elected. */
- private final CompletableFuture<AtomicReference<ClusterNode>> leaderFuture = new CompletableFuture<>();
-
- /** Container of the elected leader. */
- private final AtomicReference<ClusterNode> leaderRef = new AtomicReference<>();
-
- /** Latest lease expiration time. */
- private volatile HybridTimestamp leaseExpirationTime;
-
- /** External executor. */
- // TODO: IGNITE-20063 Maybe get rid of it
- private final ExecutorService executor;
-
- private final PlacementDriver placementDriver;
-
- private final ClockService clockService;
-
- private final BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure;
-
+public interface Replica {
/**
- * The constructor of a replica server.
+ * Returns replica's RAFT client.
*
- * @param replicaGrpId Replication group id.
- * @param listener Replica listener.
- * @param storageIndexTracker Storage index tracker.
- * @param localNode Instance of the local node.
- * @param executor External executor.
- * @param placementDriver Placement driver.
- * @param clockService Clock service.
- * @param replicaReservationClosure Closure that will be called to reserve the replica for becoming primary. It returns whether
- * the reservation was successful.
+ * @return RAFT client.
*/
- public Replica(
- ReplicationGroupId replicaGrpId,
- ReplicaListener listener,
- PendingComparableValuesTracker<Long, Void> storageIndexTracker,
- ClusterNode localNode,
- ExecutorService executor,
- PlacementDriver placementDriver,
- ClockService clockService,
- BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure
- ) {
- this.replicaGrpId = replicaGrpId;
- this.listener = listener;
- this.storageIndexTracker = storageIndexTracker;
- this.raftClient = raftClient();
- this.localNode = localNode;
- this.executor = executor;
- this.placementDriver = placementDriver;
- this.clockService = clockService;
- this.replicaReservationClosure = replicaReservationClosure;
-
- raftClient.subscribeLeader(this::onLeaderElected);
- }
-
- /** Returns Raft-client. */
- public final TopologyAwareRaftGroupService raftClient() {
- return (TopologyAwareRaftGroupService) listener.raftClient();
- }
+ TopologyAwareRaftGroupService raftClient();
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644404094
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -577,6 +640,73 @@ public CompletableFuture<Boolean> startReplica(
}
}
+ /**
+ * Starts a replica. If a replica with the same partition id already exists, the method throws an exception.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param listener Replica listener.
+ * @param snapshotStorageFactory Snapshot storage factory for raft group option's parameterization.
+ * @param newConfiguration A configuration for new raft group.
+ * @param raftGroupListener Raft group listener for raft group starting.
+ * @param raftGroupEventsListener Raft group events listener for raft group starting.
+ * @throws NodeStoppingException If node is stopping.
+ * @throws ReplicaIsAlreadyStartedException Is thrown when a replica with the same replication group id has already been
+ * started.
+ */
+ public CompletableFuture<Replica> startReplica(
Review Comment:
Nope, this method have no storageIndexTracker for example and witll not have it in future, I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644265344
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.replicator;
+
+import static java.lang.System.currentTimeMillis;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+import static org.apache.ignite.internal.util.IgniteUtils.retryOperationUntilSuccess;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
+import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
+import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
+import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaChangeCommand;
+import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
+import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Replica server.
+ */
+public class ReplicaImpl implements Replica {
Review Comment:
No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644403168
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -199,6 +200,66 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
private volatile String localNodeConsistentId;
+ /* Temporary converter to support the zone based partitions in tests. **/
+ private Function<ReplicationGroupId, ReplicationGroupId> groupIdConverter = Function.identity();
+
+ /**
+ * Constructor for a replica service.
+ *
+ * @param nodeName Node name.
+ * @param clusterNetSvc Cluster network service.
+ * @param cmgMgr Cluster group manager.
+ * @param clockService Clock service.
+ * @param messageGroupsToHandle Message handlers.
+ * @param placementDriver A placement driver.
+ * @param requestsExecutor Executor that will be used to execute requests by replicas.
+ * @param idleSafeTimePropagationPeriodMsSupplier Used to get idle safe time propagation period in ms.
+ * @param failureProcessor Failure processor.
+ * @param raftCommandsMarshaller Command marshaller for raft groups creation.
+ * @param raftGroupServiceFactory A factory for raft-clients creation.
+ * @param raftManager The manager made up of songs and words to spite all my troubles is not so bad at all.
+ * @param volatileLogStorageFactoryCreator Creator for {@link org.apache.ignite.internal.raft.storage.LogStorageFactory} for
+ * volatile tables.
+ * @param groupIdConverter Temporary converter to support the zone based partitions in tests.
+ */
+ @TestOnly
+ public ReplicaManager(
Review Comment:
Yep, made a TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1628401813
##########
modules/core/src/main/java/org/apache/ignite/internal/replicator/ZonePartitionId.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.replicator;
+
+import java.util.Objects;
+
+/**
+ * Partition id for zone based partitions.
+ */
+public class ZonePartitionId implements ReplicationGroupId {
Review Comment:
Yep, pulled the appropriate implementation from the feature branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644751568
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
Review Comment:
Not sure, that I understand, what do you mean?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644783094
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
Review Comment:
My fault, changed to reuse `PeersAndLearners.fromAssignments`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644775038
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/TimedBinaryRow.java:
##########
@@ -35,7 +35,7 @@ public TimedBinaryRow(@Nullable BinaryRow binaryRow, @Nullable HybridTimestamp c
this.commitTimestamp = commitTimestamp;
}
- TimedBinaryRow(@Nullable BinaryRow binaryRow) {
+ public TimedBinaryRow(@Nullable BinaryRow binaryRow) {
Review Comment:
Because now it is created in the table module, but placed in the partition-replicator (because it is needed for network messages).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1628403978
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
Review Comment:
Agree, done
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
Review Comment:
Yep, I think so
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1630376651
##########
modules/replica-lifecycle/build.gradle:
##########
@@ -22,9 +22,56 @@ apply from: "$rootDir/buildscripts/java-integration-test.gradle"
apply from: "$rootDir/buildscripts/java-test-fixtures.gradle"
dependencies {
Review Comment:
Renamed to partition-replica
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1645050589
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/marshaller/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.partition.replica.marshaller;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.util.VarIntUtils;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry, ByteBuffersPool cache) {
+ super(serializationRegistry, cache);
+ }
+
+ @Override
+ protected void beforeWriteMessage(Object o, ByteBuffer buffer) {
+ int requiredCatalogVersion = NO_VERSION_REQUIRED;
+
+ stream.setBuffer(buffer);
+ stream.writeInt(requiredCatalogVersion);
+ }
+
+ @Override
+ public <T> T unmarshall(ByteBuffer raw) {
+ T res = super.unmarshall(raw);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1645050303
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/marshaller/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.partition.replica.marshaller;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.util.VarIntUtils;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry, ByteBuffersPool cache) {
+ super(serializationRegistry, cache);
+ }
+
+ @Override
+ protected void beforeWriteMessage(Object o, ByteBuffer buffer) {
+ int requiredCatalogVersion = NO_VERSION_REQUIRED;
Review Comment:
I had a temporary issues with classes and modules, returned this code back, now marshaller is a copy of the same marshaller in table module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1625810743
##########
modules/core/src/main/java/org/apache/ignite/internal/replicator/ZonePartitionId.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.replicator;
+
+import java.util.Objects;
+
+/**
+ * Partition id for zone based partitions.
+ */
+public class ZonePartitionId implements ReplicationGroupId {
Review Comment:
As discussed let's reuse the one from feature branch, meaning the one with three params.
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ private final Marshaller raftCommandsMarshaller;
+
+ public static final String ZONE_STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
Review Comment:
Why not to use the one that you've introduced in utils?
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
Review Comment:
We will remove given dependency after https://issues.apache.org/jira/browse/IGNITE-21805, right?
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
Review Comment:
Let's rename it to PartitionReplicaLifecycleManager, WDYT?
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ private final Marshaller raftCommandsMarshaller;
+
+ public static final String ZONE_STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(ReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param raftMgr RAFT manager.
+ * @param replicaMgr Replica manager.
+ * @param raftGroupServiceFactory Raft clients factory.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ * @param messageSerializationRegistry Message serialization registry.
+ */
+ public ReplicaLifecycleManager(CatalogManager catalogMgr, RaftManager raftMgr, ReplicaManager replicaMgr,
Review Comment:
Could you please have them in a one at a line manner?
##########
modules/replica-lifecycle/build.gradle:
##########
@@ -22,9 +22,56 @@ apply from: "$rootDir/buildscripts/java-integration-test.gradle"
apply from: "$rootDir/buildscripts/java-test-fixtures.gradle"
dependencies {
Review Comment:
Let's also rename the module to partition-replica-lifecycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1627961179
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/ReplicaLifecycleManager.java:
##########
@@ -31,28 +89,325 @@
* - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
*/
public class ReplicaLifecycleManager implements IgniteComponent {
-
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
/* Feature flag for zone based collocation track */
// TODO IGNITE-22115 remove it
- private static final boolean ENABLED = getBoolean("IGNITE_ZONE_BASED_REPLICATION", false);
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final RaftManager raftMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final TopologyAwareRaftGroupServiceFactory raftGroupServiceFactory;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ private final Marshaller raftCommandsMarshaller;
+
+ public static final String ZONE_STABLE_ASSIGNMENTS_PREFIX = "zone.assignments.stable.";
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644775337
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/disaster/LocalPartitionStateEnum.java:
##########
@@ -42,7 +42,7 @@ public enum LocalPartitionStateEnum {
BROKEN;
/** Converts internal raft node state into public local partition state. */
- static LocalPartitionStateEnum convert(State raftNodeState) {
+ public static LocalPartitionStateEnum convert(State raftNodeState) {
Review Comment:
The same as with TimedBinaryRow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1642628111
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java:
##########
@@ -49,7 +49,7 @@
class IndexBuildTask {
private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTask.class);
- private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new TableMessagesFactory();
+ private static final PartitionReplicationMessagesFactory TABLE_MESSAGES_FACTORY = new PartitionReplicationMessagesFactory();
Review Comment:
Should we also rename "TABLE_MESSAGES_FACTORY"?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
Review Comment:
Generally I'd say that we don't want to process replica start (or in given case assignments calculation) synchronously in ms thread. It closely relates to what we've discuss with @alievmirza today. Could you please create a ticket for that and post a todo here?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionGroupListener.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.partition.replica;
+
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.raft.ReadCommand;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.service.CommandClosure;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+
+/**
+ * Listener for the zone partition.
+ */
+public class PartitionGroupListener implements RaftGroupListener {
Review Comment:
I'd name it to ZonePartitionRaftListener.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
+ var peers = new HashSet<String>();
+ var learners = new HashSet<String>();
+
+ for (Assignment assignment : assignments) {
+ if (assignment.isPeer()) {
+ peers.add(assignment.consistentId());
+ } else {
+ learners.add(assignment.consistentId());
+ }
+ }
+
+ return PeersAndLearners.fromConsistentIds(peers, learners);
+ }
+
+ private ClusterNode localNode() {
+ return topologyService.localMember();
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ for (ReplicationGroupId replicationGroupId : replicationGroupIds) {
Review Comment:
Basically, same as above. Let's consolidate the solution provided by @alievmirza with graceful tracking idea.
##########
modules/partition-replica/build.gradle:
##########
@@ -0,0 +1,78 @@
+/*
Review Comment:
I'd rename the module to "partition-replicator".
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/disaster/LocalPartitionStateEnum.java:
##########
@@ -42,7 +42,7 @@ public enum LocalPartitionStateEnum {
BROKEN;
/** Converts internal raft node state into public local partition state. */
- static LocalPartitionStateEnum convert(State raftNodeState) {
+ public static LocalPartitionStateEnum convert(State raftNodeState) {
Review Comment:
Why it's public?
##########
modules/partition-replica/src/integrationTest/java/org/apache/ignite/internal/partition/replica/utils/TestPlacementDriver.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.partition.replica.utils;
+
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.event.AbstractEventProducer;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Trivial placement driver for tests.
+ */
+public class TestPlacementDriver extends AbstractEventProducer<PrimaryReplicaEvent, PrimaryReplicaEventParameters> implements
+ PlacementDriver {
+
+ private ClusterNode primary;
Review Comment:
I'm not sure that it's guaranteed that setPrimary and await/getPrimary will be always called within same thread, thus I'd make it volatile.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
Review Comment:
AFAIK, we should place params at the end of the message.
##########
modules/partition-replica/src/integrationTest/java/org/apache/ignite/internal/partition/replica/ItReplicaLifecycleTest.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.Collections.reverse;
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.BaseIgniteRestartTest.createVault;
+import static org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_TEST_PROFILE_NAME;
+import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.REBALANCE_SCHEDULER_POOL_SIZE;
+import static org.apache.ignite.internal.partition.replica.PartitionReplicaLifecycleManager.FEATURE_FLAG_NAME;
+import static org.apache.ignite.internal.sql.SqlCommon.DEFAULT_SCHEMA_NAME;
+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.TestIgnitionManager.DEFAULT_MAX_CLOCK_SKEW_MS;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
+import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
+import static org.apache.ignite.sql.ColumnType.INT32;
+import static org.apache.ignite.sql.ColumnType.INT64;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+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.commands.ColumnParams;
+import org.apache.ignite.internal.catalog.storage.UpdateLogImpl;
+import org.apache.ignite.internal.cluster.management.ClusterInitializer;
+import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import org.apache.ignite.internal.cluster.management.NodeAttributesCollector;
+import org.apache.ignite.internal.cluster.management.configuration.ClusterManagementConfiguration;
+import org.apache.ignite.internal.cluster.management.configuration.NodeAttributesConfiguration;
+import org.apache.ignite.internal.cluster.management.raft.TestClusterStateStorage;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyImpl;
+import org.apache.ignite.internal.cluster.management.topology.LogicalTopologyServiceImpl;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
+import org.apache.ignite.internal.configuration.ConfigurationManager;
+import org.apache.ignite.internal.configuration.ConfigurationRegistry;
+import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
+import org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
+import org.apache.ignite.internal.configuration.storage.LocalFileConfigurationStorage;
+import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
+import org.apache.ignite.internal.configuration.validation.TestConfigurationValidator;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.distributionzones.DistributionZonesTestUtil;
+import org.apache.ignite.internal.failure.FailureProcessor;
+import org.apache.ignite.internal.failure.NoOpFailureProcessor;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.ClockServiceImpl;
+import org.apache.ignite.internal.hlc.ClockWaiter;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.index.IndexManager;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.lowwatermark.LowWatermarkImpl;
+import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.configuration.MetaStorageConfiguration;
+import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
+import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
+import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
+import org.apache.ignite.internal.metrics.NoOpMetricManager;
+import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.network.StaticNodeFinder;
+import org.apache.ignite.internal.network.configuration.NetworkConfiguration;
+import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
+import org.apache.ignite.internal.pagememory.configuration.schema.PersistentPageMemoryProfileConfigurationSchema;
+import org.apache.ignite.internal.pagememory.configuration.schema.UnsafeMemoryAllocatorConfigurationSchema;
+import org.apache.ignite.internal.pagememory.configuration.schema.VolatilePageMemoryProfileConfigurationSchema;
+import org.apache.ignite.internal.partition.replica.network.PartitionReplicationMessageGroup;
+import org.apache.ignite.internal.partition.replica.utils.TestPlacementDriver;
+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.raft.storage.LogStorageFactory;
+import org.apache.ignite.internal.raft.storage.impl.LocalLogStorageFactory;
+import org.apache.ignite.internal.raft.util.SharedLogStorageFactoryUtils;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.replicator.configuration.ReplicationConfiguration;
+import org.apache.ignite.internal.schema.SchemaManager;
+import org.apache.ignite.internal.schema.configuration.GcConfiguration;
+import org.apache.ignite.internal.schema.configuration.StorageUpdateConfiguration;
+import org.apache.ignite.internal.storage.DataStorageManager;
+import org.apache.ignite.internal.storage.DataStorageModules;
+import org.apache.ignite.internal.storage.configurations.StorageConfiguration;
+import org.apache.ignite.internal.storage.pagememory.PersistentPageMemoryDataStorageModule;
+import org.apache.ignite.internal.storage.pagememory.configuration.schema.PersistentPageMemoryStorageEngineExtensionConfigurationSchema;
+import org.apache.ignite.internal.storage.pagememory.configuration.schema.VolatilePageMemoryStorageEngineExtensionConfigurationSchema;
+import org.apache.ignite.internal.table.TableTestUtils;
+import org.apache.ignite.internal.table.distributed.TableManager;
+import org.apache.ignite.internal.table.distributed.raft.snapshot.outgoing.OutgoingSnapshotsManager;
+import org.apache.ignite.internal.table.distributed.schema.SchemaSyncService;
+import org.apache.ignite.internal.table.distributed.schema.SchemaSyncServiceImpl;
+import org.apache.ignite.internal.table.distributed.schema.ThreadLocalPartitionCommandsMarshaller;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.tx.HybridTimestampTracker;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.configuration.TransactionConfiguration;
+import org.apache.ignite.internal.tx.impl.HeapLockManager;
+import org.apache.ignite.internal.tx.impl.RemotelyTriggeredResourceRegistry;
+import org.apache.ignite.internal.tx.impl.TransactionIdGenerator;
+import org.apache.ignite.internal.tx.impl.TransactionInflights;
+import org.apache.ignite.internal.tx.impl.TxManagerImpl;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.test.TestLocalRwTxCounter;
+import org.apache.ignite.internal.vault.VaultManager;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.raft.jraft.rpc.impl.RaftGroupEventsClientListener;
+import org.apache.ignite.sql.IgniteSql;
+import org.apache.ignite.table.KeyValueView;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Replica lifecycle test.
+ */
+@ExtendWith({WorkDirectoryExtension.class, ConfigurationExtension.class})
+@Timeout(60)
+public class ItReplicaLifecycleTest extends BaseIgniteAbstractTest {
Review Comment:
Let's create a special ticket for feature GC and left bunch of todos here and there marking the code that should be removed on feature-branch merge. E.g. given test class should be eventually removed in order not to forget it let's mark it with corresponding todo.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/PartitionReplicationMessageGroup.java:
##########
@@ -15,52 +15,51 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.table.distributed;
+package org.apache.ignite.internal.partition.replica.network;
-import static org.apache.ignite.internal.table.distributed.TableMessageGroup.GROUP_TYPE;
+import static org.apache.ignite.internal.partition.replica.network.PartitionReplicationMessageGroup.GROUP_TYPE;
import org.apache.ignite.internal.network.annotations.MessageGroup;
-import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
-import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
-import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
-import org.apache.ignite.internal.table.distributed.command.TimedBinaryRowMessage;
-import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
-import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
-import org.apache.ignite.internal.table.distributed.command.WriteIntentSwitchCommand;
-import org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStateMessage;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStatesRequest;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStatesResponse;
-import org.apache.ignite.internal.table.distributed.message.HasDataRequest;
-import org.apache.ignite.internal.table.distributed.message.HasDataResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMetaRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMetaResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataResponse.ResponseEntry;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotTxDataRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotTxDataResponse;
-import org.apache.ignite.internal.table.distributed.replication.request.BinaryRowMessage;
-import org.apache.ignite.internal.table.distributed.replication.request.BinaryTupleMessage;
-import org.apache.ignite.internal.table.distributed.replication.request.BuildIndexReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyDirectMultiRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyDirectSingleRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyMultiRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyScanRetrieveBatchReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlySingleRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteMultiRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteMultiRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteScanRetrieveBatchReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSwapRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ScanCloseReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.command.BuildIndexCommand;
+import org.apache.ignite.internal.partition.replica.network.command.FinishTxCommand;
+import org.apache.ignite.internal.partition.replica.network.command.TablePartitionIdMessage;
+import org.apache.ignite.internal.partition.replica.network.command.TimedBinaryRowMessage;
+import org.apache.ignite.internal.partition.replica.network.command.UpdateAllCommand;
+import org.apache.ignite.internal.partition.replica.network.command.UpdateCommand;
+import org.apache.ignite.internal.partition.replica.network.command.WriteIntentSwitchCommand;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStateMessage;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStatesRequest;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStatesResponse;
+import org.apache.ignite.internal.partition.replica.network.message.HasDataRequest;
+import org.apache.ignite.internal.partition.replica.network.message.HasDataResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMetaRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMetaResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataResponse.ResponseEntry;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotTxDataRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotTxDataResponse;
+import org.apache.ignite.internal.partition.replica.network.replication.BinaryRowMessage;
+import org.apache.ignite.internal.partition.replica.network.replication.BinaryTupleMessage;
+import org.apache.ignite.internal.partition.replica.network.replication.BuildIndexReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyDirectMultiRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyDirectSingleRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyMultiRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyScanRetrieveBatchReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlySingleRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteMultiRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteMultiRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteScanRetrieveBatchReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSingleRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSingleRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSwapRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ScanCloseReplicaRequest;
/**
* Message group for the table module.
*/
-@MessageGroup(groupType = GROUP_TYPE, groupName = "TableMessages")
-public interface TableMessageGroup {
+@MessageGroup(groupType = GROUP_TYPE, groupName = "PartitionReplicationMessages")
+public interface PartitionReplicationMessageGroup {
/** Table message group type. */
Review Comment:
Table.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
Review Comment:
One at a line please.
##########
modules/partition-replica/src/integrationTest/java/org/apache/ignite/internal/partition/replica/utils/TestPlacementDriver.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * 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.partition.replica.utils;
+
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.event.AbstractEventProducer;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Trivial placement driver for tests.
+ */
+public class TestPlacementDriver extends AbstractEventProducer<PrimaryReplicaEvent, PrimaryReplicaEventParameters> implements
Review Comment:
Same as above. Let's add removal todo here.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionGroupListener.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.partition.replica;
+
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.raft.ReadCommand;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.service.CommandClosure;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+
+/**
+ * Listener for the zone partition.
Review Comment:
I'd mention that it's raft listener.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
Review Comment:
`createZoneReplicationEntities` and `createZonePartitionReplicationNodes` should clearly state through naming that first belong to partition**s** and second to partition. I'd rather use consistent pattern related to Entities/Nodes etc.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
Review Comment:
Unnecessary empty line.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
Review Comment:
Seems that it's not partition specific. Should we move it to replicator?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
Review Comment:
Unnecessary empty line.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
Review Comment:
busy lock is missing. Despite the fact that we skip it widely, seems that it's easy to add it here, so let's do it.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
Review Comment:
Should we just complete the future exceptionally? Not only in case of NodeStoppingException.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
Review Comment:
Let's encapsulate it to should start locally, WDYT?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/marshaller/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.partition.replica.marshaller;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.util.VarIntUtils;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry, ByteBuffersPool cache) {
+ super(serializationRegistry, cache);
+ }
+
+ @Override
+ protected void beforeWriteMessage(Object o, ByteBuffer buffer) {
+ int requiredCatalogVersion = NO_VERSION_REQUIRED;
Review Comment:
Why there's no
```
int requiredCatalogVersion = o instanceof CatalogVersionAware
? ((CatalogVersionAware) o).requiredCatalogVersion()
: NO_VERSION_REQUIRED;
```
?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -199,6 +200,66 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
private volatile String localNodeConsistentId;
+ /* Temporary converter to support the zone based partitions in tests. **/
+ private Function<ReplicationGroupId, ReplicationGroupId> groupIdConverter = Function.identity();
+
+ /**
+ * Constructor for a replica service.
+ *
+ * @param nodeName Node name.
+ * @param clusterNetSvc Cluster network service.
+ * @param cmgMgr Cluster group manager.
+ * @param clockService Clock service.
+ * @param messageGroupsToHandle Message handlers.
+ * @param placementDriver A placement driver.
+ * @param requestsExecutor Executor that will be used to execute requests by replicas.
+ * @param idleSafeTimePropagationPeriodMsSupplier Used to get idle safe time propagation period in ms.
+ * @param failureProcessor Failure processor.
+ * @param raftCommandsMarshaller Command marshaller for raft groups creation.
+ * @param raftGroupServiceFactory A factory for raft-clients creation.
+ * @param raftManager The manager made up of songs and words to spite all my troubles is not so bad at all.
+ * @param volatileLogStorageFactoryCreator Creator for {@link org.apache.ignite.internal.raft.storage.LogStorageFactory} for
+ * volatile tables.
+ * @param groupIdConverter Temporary converter to support the zone based partitions in tests.
+ */
+ @TestOnly
+ public ReplicaManager(
Review Comment:
Is it expected to be removed after full feature merge?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/PartitionReplicationMessageGroup.java:
##########
@@ -15,52 +15,51 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.table.distributed;
+package org.apache.ignite.internal.partition.replica.network;
-import static org.apache.ignite.internal.table.distributed.TableMessageGroup.GROUP_TYPE;
+import static org.apache.ignite.internal.partition.replica.network.PartitionReplicationMessageGroup.GROUP_TYPE;
import org.apache.ignite.internal.network.annotations.MessageGroup;
-import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
-import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
-import org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
-import org.apache.ignite.internal.table.distributed.command.TimedBinaryRowMessage;
-import org.apache.ignite.internal.table.distributed.command.UpdateAllCommand;
-import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
-import org.apache.ignite.internal.table.distributed.command.WriteIntentSwitchCommand;
-import org.apache.ignite.internal.table.distributed.disaster.DisasterRecoveryManager;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStateMessage;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStatesRequest;
-import org.apache.ignite.internal.table.distributed.disaster.messages.LocalPartitionStatesResponse;
-import org.apache.ignite.internal.table.distributed.message.HasDataRequest;
-import org.apache.ignite.internal.table.distributed.message.HasDataResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMetaRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMetaResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataResponse;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotMvDataResponse.ResponseEntry;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotTxDataRequest;
-import org.apache.ignite.internal.table.distributed.raft.snapshot.message.SnapshotTxDataResponse;
-import org.apache.ignite.internal.table.distributed.replication.request.BinaryRowMessage;
-import org.apache.ignite.internal.table.distributed.replication.request.BinaryTupleMessage;
-import org.apache.ignite.internal.table.distributed.replication.request.BuildIndexReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyDirectMultiRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyDirectSingleRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyMultiRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlyScanRetrieveBatchReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadOnlySingleRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteMultiRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteMultiRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteScanRetrieveBatchReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowPkReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSingleRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ReadWriteSwapRowReplicaRequest;
-import org.apache.ignite.internal.table.distributed.replication.request.ScanCloseReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.command.BuildIndexCommand;
+import org.apache.ignite.internal.partition.replica.network.command.FinishTxCommand;
+import org.apache.ignite.internal.partition.replica.network.command.TablePartitionIdMessage;
+import org.apache.ignite.internal.partition.replica.network.command.TimedBinaryRowMessage;
+import org.apache.ignite.internal.partition.replica.network.command.UpdateAllCommand;
+import org.apache.ignite.internal.partition.replica.network.command.UpdateCommand;
+import org.apache.ignite.internal.partition.replica.network.command.WriteIntentSwitchCommand;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStateMessage;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStatesRequest;
+import org.apache.ignite.internal.partition.replica.network.disaster.LocalPartitionStatesResponse;
+import org.apache.ignite.internal.partition.replica.network.message.HasDataRequest;
+import org.apache.ignite.internal.partition.replica.network.message.HasDataResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMetaRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMetaResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataResponse;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotMvDataResponse.ResponseEntry;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotTxDataRequest;
+import org.apache.ignite.internal.partition.replica.network.raft.SnapshotTxDataResponse;
+import org.apache.ignite.internal.partition.replica.network.replication.BinaryRowMessage;
+import org.apache.ignite.internal.partition.replica.network.replication.BinaryTupleMessage;
+import org.apache.ignite.internal.partition.replica.network.replication.BuildIndexReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyDirectMultiRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyDirectSingleRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyMultiRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlyScanRetrieveBatchReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadOnlySingleRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteMultiRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteMultiRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteScanRetrieveBatchReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSingleRowPkReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSingleRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ReadWriteSwapRowReplicaRequest;
+import org.apache.ignite.internal.partition.replica.network.replication.ScanCloseReplicaRequest;
/**
* Message group for the table module.
*/
-@MessageGroup(groupType = GROUP_TYPE, groupName = "TableMessages")
-public interface TableMessageGroup {
+@MessageGroup(groupType = GROUP_TYPE, groupName = "PartitionReplicationMessages")
+public interface PartitionReplicationMessageGroup {
/** Table message group type. */
short GROUP_TYPE = 9;
Review Comment:
There are no messages in table module any longer? Is it true for GG9?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/network/TimedBinaryRow.java:
##########
@@ -35,7 +35,7 @@ public TimedBinaryRow(@Nullable BinaryRow binaryRow, @Nullable HybridTimestamp c
this.commitTimestamp = commitTimestamp;
}
- TimedBinaryRow(@Nullable BinaryRow binaryRow) {
+ public TimedBinaryRow(@Nullable BinaryRow binaryRow) {
Review Comment:
Why it's public?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
+ var peers = new HashSet<String>();
+ var learners = new HashSet<String>();
+
+ for (Assignment assignment : assignments) {
+ if (assignment.isPeer()) {
+ peers.add(assignment.consistentId());
+ } else {
+ learners.add(assignment.consistentId());
+ }
+ }
+
+ return PeersAndLearners.fromConsistentIds(peers, learners);
+ }
+
+ private ClusterNode localNode() {
+ return topologyService.localMember();
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ for (ReplicationGroupId replicationGroupId : replicationGroupIds) {
+ try {
+ replicaMgr.stopReplica(replicationGroupId);
+ } catch (NodeStoppingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * 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 zoneId Zone id.
+ * @param assignmentsFuture Assignments future, to get the assignments that should be written.
+ * @return Real list of assignments.
+ */
+ private CompletableFuture<List<Assignments>> writeTableAssignmentsToMetastore(
Review Comment:
Table?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/ZonePartitionReplicaListener.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.partition.replica;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
+import org.apache.ignite.internal.raft.service.RaftCommandRunner;
+import org.apache.ignite.internal.replicator.ReplicaResult;
+import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.schema.BinaryRowImpl;
+
+/**
+ * Zone partition replica listener.
+ */
+public class ZonePartitionReplicaListener implements ReplicaListener {
+
+ @Override
+ public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, String senderId) {
+ var res =
Review Comment:
Formatting is a bit strange. Never mind, it's updated in your next PR.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java:
##########
@@ -17,126 +17,22 @@
package org.apache.ignite.internal.replicator;
-import static java.lang.System.currentTimeMillis;
-import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
-import static org.apache.ignite.internal.util.IgniteUtils.retryOperationUntilSuccess;
-
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiFunction;
-import org.apache.ignite.internal.hlc.ClockService;
-import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.lang.IgniteStringFormatter;
-import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.NetworkMessage;
-import org.apache.ignite.internal.placementdriver.PlacementDriver;
-import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
-import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
-import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
-import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
-import org.apache.ignite.internal.replicator.listener.ReplicaListener;
-import org.apache.ignite.internal.replicator.message.PrimaryReplicaChangeCommand;
-import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaRequest;
-import org.apache.ignite.internal.util.PendingComparableValuesTracker;
-import org.apache.ignite.network.ClusterNode;
/**
- * Replica server.
+ * Interface for replica server.
*/
-public class Replica {
- /** The logger. */
- private static final IgniteLogger LOG = Loggers.forClass(ReplicaManager.class);
-
- /** Message factory. */
- private static final PlacementDriverMessagesFactory PLACEMENT_DRIVER_MESSAGES_FACTORY = new PlacementDriverMessagesFactory();
-
- private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
-
- /** Replica group identity, this id is the same as the considered partition's id. */
- private final ReplicationGroupId replicaGrpId;
-
- /** Replica listener. */
- private final ReplicaListener listener;
-
- /** Storage index tracker. */
- private final PendingComparableValuesTracker<Long, Void> storageIndexTracker;
-
- /** Topology aware Raft client. */
- private final TopologyAwareRaftGroupService raftClient;
-
- /** Instance of the local node. */
- private final ClusterNode localNode;
-
- // TODO IGNITE-19120 after replica inoperability logic is introduced, this future should be replaced with something like
- // VersionedValue (so that PlacementDriverMessages would wait for new leader election)
- /** Completes when leader is elected. */
- private final CompletableFuture<AtomicReference<ClusterNode>> leaderFuture = new CompletableFuture<>();
-
- /** Container of the elected leader. */
- private final AtomicReference<ClusterNode> leaderRef = new AtomicReference<>();
-
- /** Latest lease expiration time. */
- private volatile HybridTimestamp leaseExpirationTime;
-
- /** External executor. */
- // TODO: IGNITE-20063 Maybe get rid of it
- private final ExecutorService executor;
-
- private final PlacementDriver placementDriver;
-
- private final ClockService clockService;
-
- private final BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure;
-
+public interface Replica {
/**
- * The constructor of a replica server.
+ * Returns replica's RAFT client.
*
- * @param replicaGrpId Replication group id.
- * @param listener Replica listener.
- * @param storageIndexTracker Storage index tracker.
- * @param localNode Instance of the local node.
- * @param executor External executor.
- * @param placementDriver Placement driver.
- * @param clockService Clock service.
- * @param replicaReservationClosure Closure that will be called to reserve the replica for becoming primary. It returns whether
- * the reservation was successful.
+ * @return RAFT client.
*/
- public Replica(
- ReplicationGroupId replicaGrpId,
- ReplicaListener listener,
- PendingComparableValuesTracker<Long, Void> storageIndexTracker,
- ClusterNode localNode,
- ExecutorService executor,
- PlacementDriver placementDriver,
- ClockService clockService,
- BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure
- ) {
- this.replicaGrpId = replicaGrpId;
- this.listener = listener;
- this.storageIndexTracker = storageIndexTracker;
- this.raftClient = raftClient();
- this.localNode = localNode;
- this.executor = executor;
- this.placementDriver = placementDriver;
- this.clockService = clockService;
- this.replicaReservationClosure = replicaReservationClosure;
-
- raftClient.subscribeLeader(this::onLeaderElected);
- }
-
- /** Returns Raft-client. */
- public final TopologyAwareRaftGroupService raftClient() {
- return (TopologyAwareRaftGroupService) listener.raftClient();
- }
+ TopologyAwareRaftGroupService raftClient();
Review Comment:
I'd expect given method to be @Deprecated(forRemoval=true) with todo linked to some of @JAkutenshi tickets.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -154,6 +156,47 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
private String localNodeId;
+ private Function<ReplicationGroupId, ReplicationGroupId> groupIdConverter = Function.identity();
Review Comment:
Since it's a temporary code, let's mark it with our common todo for removal link.
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionReplicaLifecycleManager.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.partition.replica;
+
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.ZONE_CREATE;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.stablePartAssignmentsKey;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zoneAssignmentsGetLocally;
+import static org.apache.ignite.internal.distributionzones.rebalance.RebalanceUtil.zonePartitionAssignmentsGetLocally;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.getBoolean;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+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.ConcurrentHashMap;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.affinity.AffinityUtils;
+import org.apache.ignite.internal.affinity.Assignment;
+import org.apache.ignite.internal.affinity.Assignments;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateZoneEventParameters;
+import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.ComponentContext;
+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.dsl.Condition;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.network.TopologyService;
+import org.apache.ignite.internal.partition.replica.snapshot.FailFastSnapshotStorageFactory;
+import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.RaftGroupEventsListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.replicator.ReplicaManager;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * The main responsibilities of this class:
+ * - Start the appropriate replication nodes on the zone creation.
+ * - Stop the same entities on the zone removing.
+ * - Support the rebalance mechanism and start the new replication entities when the rebalance triggers occurred.
+ */
+public class PartitionReplicaLifecycleManager implements IgniteComponent {
+ public static final String FEATURE_FLAG_NAME = "IGNITE_ZONE_BASED_REPLICATION";
+ /* Feature flag for zone based collocation track */
+ // TODO IGNITE-22115 remove it
+ public static final boolean ENABLED = getBoolean(FEATURE_FLAG_NAME, false);
+
+ private final CatalogManager catalogMgr;
+
+ private final ReplicaManager replicaMgr;
+
+ private final DistributionZoneManager distributionZoneMgr;
+
+ private final MetaStorageManager metaStorageMgr;
+
+ private final TopologyService topologyService;
+
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(PartitionReplicaLifecycleManager.class);
+
+ private final Set<ReplicationGroupId> replicationGroupIds = ConcurrentHashMap.newKeySet();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /**
+ * The constructor.
+ *
+ * @param catalogMgr Catalog manager.
+ * @param replicaMgr Replica manager.
+ * @param distributionZoneMgr Distribution zone manager.
+ * @param metaStorageMgr Metastorage manager.
+ * @param topologyService Topology service.
+ */
+ public PartitionReplicaLifecycleManager(CatalogManager catalogMgr,
+ ReplicaManager replicaMgr,
+ DistributionZoneManager distributionZoneMgr,
+ MetaStorageManager metaStorageMgr,
+ TopologyService topologyService
+ ) {
+ this.catalogMgr = catalogMgr;
+ this.replicaMgr = replicaMgr;
+ this.distributionZoneMgr = distributionZoneMgr;
+ this.metaStorageMgr = metaStorageMgr;
+ this.topologyService = topologyService;
+ }
+
+ @Override
+ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
+ if (!ENABLED) {
+ return nullCompletedFuture();
+ }
+
+ catalogMgr.listen(ZONE_CREATE,
+ (CreateZoneEventParameters parameters) ->
+ inBusyLock(busyLock, () -> onCreateZone(parameters).thenApply((ignored) -> false))
+ );
+
+ return nullCompletedFuture();
+ }
+
+ private CompletableFuture<Void> onCreateZone(CreateZoneEventParameters createZoneEventParameters) {
+
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<List<Assignments>> assignmentsFuture = getOrCreateAssignments(
+ createZoneEventParameters.zoneDescriptor(),
+ createZoneEventParameters.causalityToken(),
+ createZoneEventParameters.catalogVersion()
+ );
+
+ CompletableFuture<List<Assignments>> assignmentsFutureAfterInvoke =
+ writeTableAssignmentsToMetastore(createZoneEventParameters.zoneDescriptor().id(), assignmentsFuture);
+
+ return createZoneReplicationEntities(assignmentsFutureAfterInvoke, createZoneEventParameters.zoneDescriptor().id());
+ });
+ }
+
+ private CompletableFuture<Void> createZoneReplicationEntities(CompletableFuture<List<Assignments>> assignmentsFuture, int zoneId) {
+ return assignmentsFuture.thenCompose(assignments -> {
+ assert assignments != null : IgniteStringFormatter.format("Zone [id={}] has empty assignments.", zoneId);
+
+ List<CompletableFuture<?>> partitionsStartFutures = new ArrayList<>();
+
+ for (int i = 0; i < assignments.size(); i++) {
+ int partId = i;
+
+ partitionsStartFutures.add(createZonePartitionReplicationNodes(zoneId, partId, assignments.get(i)));
+ }
+
+ return allOf(partitionsStartFutures.toArray(new CompletableFuture<?>[0]));
+ });
+ }
+
+ private CompletableFuture<Void> createZonePartitionReplicationNodes(int zoneId, int partId, Assignments assignments) {
+
+ Assignment localMemberAssignment = assignments.nodes().stream()
+ .filter(a -> a.consistentId().equals(localNode().name()))
+ .findAny()
+ .orElse(null);
+
+ if (localMemberAssignment == null) {
+ return nullCompletedFuture();
+ }
+
+ PeersAndLearners realConfiguration = configurationFromAssignments(assignments.nodes());
+
+ ZonePartitionId replicaGrpId = new ZonePartitionId(zoneId, partId);
+
+ RaftGroupListener raftGroupListener = new PartitionGroupListener();
+
+ try {
+ return replicaMgr.startReplica(
+ replicaGrpId,
+ new ZonePartitionReplicaListener(),
+ new FailFastSnapshotStorageFactory(),
+ realConfiguration,
+ raftGroupListener,
+ RaftGroupEventsListener.noopLsnr
+ ).thenRun(() -> replicationGroupIds.add(replicaGrpId));
+ } catch (NodeStoppingException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ private static PeersAndLearners configurationFromAssignments(Collection<Assignment> assignments) {
+ var peers = new HashSet<String>();
+ var learners = new HashSet<String>();
+
+ for (Assignment assignment : assignments) {
+ if (assignment.isPeer()) {
+ peers.add(assignment.consistentId());
+ } else {
+ learners.add(assignment.consistentId());
+ }
+ }
+
+ return PeersAndLearners.fromConsistentIds(peers, learners);
+ }
+
+ private ClusterNode localNode() {
+ return topologyService.localMember();
+ }
+
+ @Override
+ public void beforeNodeStop() {
+ for (ReplicationGroupId replicationGroupId : replicationGroupIds) {
+ try {
+ replicaMgr.stopReplica(replicationGroupId);
+ } catch (NodeStoppingException e) {
Review Comment:
I'd say that we should **proceed** stopping replicas and/or complete pending replica futures with exceptions on exception on node stop.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaImpl.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.replicator;
+
+import static java.lang.System.currentTimeMillis;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+import static org.apache.ignite.internal.util.IgniteUtils.retryOperationUntilSuccess;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessage;
+import org.apache.ignite.internal.placementdriver.message.LeaseGrantedMessageResponse;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverMessagesFactory;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
+import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
+import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaChangeCommand;
+import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
+import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Replica server.
+ */
+public class ReplicaImpl implements Replica {
Review Comment:
Are there any real changes here?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -577,6 +640,73 @@ public CompletableFuture<Boolean> startReplica(
}
}
+ /**
+ * Starts a replica. If a replica with the same partition id already exists, the method throws an exception.
+ *
+ * @param replicaGrpId Replication group id.
+ * @param listener Replica listener.
+ * @param snapshotStorageFactory Snapshot storage factory for raft group option's parameterization.
+ * @param newConfiguration A configuration for new raft group.
+ * @param raftGroupListener Raft group listener for raft group starting.
+ * @param raftGroupEventsListener Raft group events listener for raft group starting.
+ * @throws NodeStoppingException If node is stopping.
+ * @throws ReplicaIsAlreadyStartedException Is thrown when a replica with the same replication group id has already been
+ * started.
+ */
+ public CompletableFuture<Replica> startReplica(
Review Comment:
Is it a merge artifact? Why do we need this additional method to ones introduces by Misha?
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/marshaller/PartitionCommandsMarshallerImpl.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.partition.replica.marshaller;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.internal.raft.util.OptimizedMarshaller;
+import org.apache.ignite.internal.util.VarIntUtils;
+
+/**
+ * Default {@link PartitionCommandsMarshaller} implementation.
+ */
+public class PartitionCommandsMarshallerImpl extends OptimizedMarshaller implements PartitionCommandsMarshaller {
+ public PartitionCommandsMarshallerImpl(MessageSerializationRegistry serializationRegistry, ByteBuffersPool cache) {
+ super(serializationRegistry, cache);
+ }
+
+ @Override
+ protected void beforeWriteMessage(Object o, ByteBuffer buffer) {
+ int requiredCatalogVersion = NO_VERSION_REQUIRED;
+
+ stream.setBuffer(buffer);
+ stream.writeInt(requiredCatalogVersion);
+ }
+
+ @Override
+ public <T> T unmarshall(ByteBuffer raw) {
+ T res = super.unmarshall(raw);
Review Comment:
Same here, why it's not aware of catalog version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1644624260
##########
modules/partition-replica/src/main/java/org/apache/ignite/internal/partition/replica/PartitionGroupListener.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.partition.replica;
+
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.raft.ReadCommand;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.raft.service.CommandClosure;
+import org.apache.ignite.internal.raft.service.RaftGroupListener;
+
+/**
+ * Listener for the zone partition.
+ */
+public class PartitionGroupListener implements RaftGroupListener {
Review Comment:
Seems that it's still PartitionGroupListener.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1628398198
##########
modules/replicator/src/main/java/org/apache/ignite/internal/ZonePartitionReplica.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.placementdriver.message.PlacementDriverReplicaMessage;
+import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
+import org.apache.ignite.internal.replicator.Replica;
+import org.apache.ignite.internal.replicator.ReplicaResult;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.jetbrains.annotations.TestOnly;
+
+/**
+ * Replica for the zone based partitions.
+ */
+public class ZonePartitionReplica extends Replica {
Review Comment:
I think it won't be removed, it will the only replica instead. Current Replica will be transformed to the another abstraction, I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22231 Implement zone based replicas start on the zone creation [ignite-3]
Posted by "kgusakov (via GitHub)" <gi...@apache.org>.
kgusakov commented on code in PR #3858:
URL: https://github.com/apache/ignite-3/pull/3858#discussion_r1628407738
##########
modules/replica-lifecycle/src/main/java/org/apache/ignite/internal/datareplication/snapshot/FailFastSnapshotStorageFactory.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.datareplication.snapshot;
+
+import org.apache.ignite.internal.raft.storage.SnapshotStorageFactory;
+import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.option.SnapshotCopierOptions;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
+import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotCopier;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
+
+/**
+ * Temporary snapshot factory, which must help to identify the unexpected snapshots,
+ * while zone based replicas is not support snapshotting yet.
+ */
+// TODO https://issues.apache.org/jira/browse/IGNITE-22115 remove it
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org