sanpwc commented on code in PR #927: URL: https://github.com/apache/ignite-3/pull/927#discussion_r929914317
########## modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java: ########## @@ -0,0 +1,200 @@ +/* + * 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.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.replicator.exception.ReplicaAlreadyIsStartedException; +import org.apache.ignite.internal.replicator.message.ReplicaMessageGroup; +import org.apache.ignite.internal.replicator.message.ReplicaRequest; +import org.apache.ignite.internal.replicator.message.ReplicaResponse; +import org.apache.ignite.internal.util.IgniteSpinBusyLock; +import org.apache.ignite.lang.IgniteException; +import org.apache.ignite.lang.IgniteStringFormatter; +import org.apache.ignite.lang.NodeStoppingException; +import org.apache.ignite.network.ClusterService; +import org.apache.ignite.network.NetworkAddress; +import org.apache.ignite.network.NetworkMessage; +import org.apache.ignite.network.NetworkMessageHandler; +import org.jetbrains.annotations.Nullable; + +/** + * Manager to rule replicas. + * Only a single instance of the class exists in Ignite node. + * This class allow to start/stop/get a replica. + */ +public class ReplicaManager implements IgniteComponent { + /** Busy lock to stop synchronously. */ + private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock(); + + /** Prevents double stopping the component. */ + private final AtomicBoolean stopGuard = new AtomicBoolean(); + + /** Cluster network service. */ + private final ClusterService clusterNetSvc; + + /** Replicas. */ + private final ConcurrentHashMap<String, Replica> replicas = new ConcurrentHashMap<>(); + + /** + * Constructor for replica service. + * + * @param clusterNetSvc Cluster network service. + */ + public ReplicaManager(ClusterService clusterNetSvc) { + this.clusterNetSvc = clusterNetSvc; + } + + /** + * Gets a replica by its replica group id. + * + * @param replicaGrpId Replication group id. + * @return Instance of the replica or {@code null} if the replica is not started. + * @throws NodeStoppingException Is thrown when the node is stopping. + */ + public Replica replica(String replicaGrpId) throws NodeStoppingException { + if (!busyLock.enterBusy()) { + throw new NodeStoppingException(); + } + + try { + return replicas.get(replicaGrpId); + } finally { + busyLock.leaveBusy(); + } + } + + /** + * Starts a replica. If a replica with the same partition id is already exist the method throws an exception. + * + * @param replicaGrpId Replication group id. + * @param listener Replica listener. + * @return Replica. + * @throws NodeStoppingException Is thrown when the node is stopping. + * @throws ReplicaAlreadyIsStartedException Is thrown when a replica with the same replication group id already started. + */ + public Replica startReplica(String replicaGrpId, ReplicaListener listener) throws NodeStoppingException { + if (!busyLock.enterBusy()) { + throw new NodeStoppingException(); + } + + try { + return startReplicaInternal(replicaGrpId, listener); + } finally { + busyLock.leaveBusy(); + } + } + + /** + * Internal method for start a replica. + * + * @param replicaGrpId Replication group id. + * @param listener Replica listener. + * @return Replica. + */ + private Replica startReplicaInternal(String replicaGrpId, ReplicaListener listener) { + var replica = new Replica(replicaGrpId, listener); + + Replica previous = replicas.putIfAbsent(replicaGrpId, replica); + + if (previous != null) { + throw new ReplicaAlreadyIsStartedException(replicaGrpId); + } + + return replica; + } + + /** + * Stops a replica by the partition group id. + * + * @param replicaGrpId Replication group id. + * @return True if the replica is found and closed, false otherwise. + * @throws NodeStoppingException Is thrown when the node is stopping. + */ + public boolean stopReplica(String replicaGrpId) throws NodeStoppingException { + if (!busyLock.enterBusy()) { + throw new NodeStoppingException(); + } + + try { + return stopReplicaInternal(replicaGrpId); + } finally { + busyLock.leaveBusy(); + } + } + + /** + * An internal method for stopping a replica. + * + * @param replicaGrpId Replication group id. + * @return True if the replica is found and closed, false otherwise. + */ + private boolean stopReplicaInternal(String replicaGrpId) { + Replica replica = replicas.remove(replicaGrpId); + + if (replica == null) { + return false; + } + + return true; + } + + /** {@inheritDoc} */ + @Override + public void start() { + clusterNetSvc.messagingService().addMessageHandler(ReplicaMessageGroup.class, new NetworkMessageHandler() { + @Override + public void onReceived(NetworkMessage message, NetworkAddress senderAddr, @Nullable Long correlationId) { + if (!busyLock.enterBusy()) { + throw new IgniteException(new NodeStoppingException()); + } + + try { + assert message instanceof ReplicaRequest : IgniteStringFormatter.format("Unexpected message [message={}]", message); + + ReplicaRequest request = (ReplicaRequest) message; + + Replica replica = replicas.get(request.locator().groupId()); + + if (replica == null) { + //TODO:IGNITE-17255 Sent an exceptional response to the client side when the replica is absent. Review Comment: sen**d** -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
