You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/12/08 08:04:33 UTC
[GitHub] [ignite-3] rpuch commented on a diff in pull request #1397: IGNITE-18264 Add Peer index support
rpuch commented on code in PR #1397:
URL: https://github.com/apache/ignite-3/pull/1397#discussion_r1042073284
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/PeersAndLearners.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.raft;
+
+import static java.util.stream.Collectors.toUnmodifiableSet;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Class containing peers and learners of a Raft Group.
+ */
+public class PeersAndLearners {
+ @IgniteToStringInclude
+ private final Set<Peer> peers;
+
+ @IgniteToStringInclude
+ private final Set<Peer> learners;
+
+ private PeersAndLearners(Collection<Peer> peers, Collection<Peer> learners) {
+ this.peers = Set.copyOf(peers);
+ this.learners = Set.copyOf(learners);
+ }
+
+ /**
+ * Creates an instance using peers represented as their consistent IDs.
+ */
+ public static PeersAndLearners fromConsistentIds(Set<String> peerNames) {
+ return fromConsistentIds(peerNames, Set.of());
+ }
+
+ /**
+ * Creates an instance using peers and learners represented as their consistent IDs.
+ */
+ public static PeersAndLearners fromConsistentIds(Set<String> peerNames, Set<String> learnerNames) {
+ Set<Peer> peers = peerNames.stream().map(Peer::new).collect(toUnmodifiableSet());
+
+ Set<Peer> learners = learnerNames.stream()
+ .map(name -> {
+ int idx = peerNames.contains(name) ? 1 : 0;
+
+ return new Peer(name, idx);
+ })
+ .collect(toUnmodifiableSet());
+
+ return new PeersAndLearners(peers, learners);
+ }
+
+ /**
+ * Creates an instance using peers and learners represented as {@link Peer}s.
+ */
+ public static PeersAndLearners fromPeers(Collection<Peer> peers, Collection<Peer> learners) {
+ assert Collections.disjoint(peers, learners);
Review Comment:
`Peer#equals()` compares using `consistentId`, `index` AND `priority`, so we might have a peer and a learner with same `consistentId`+`index`, but different priorities. Is this ok?
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/PeersAndLearners.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.raft;
+
+import static java.util.stream.Collectors.toUnmodifiableSet;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Class containing peers and learners of a Raft Group.
+ */
+public class PeersAndLearners {
+ @IgniteToStringInclude
+ private final Set<Peer> peers;
+
+ @IgniteToStringInclude
+ private final Set<Peer> learners;
+
+ private PeersAndLearners(Collection<Peer> peers, Collection<Peer> learners) {
+ this.peers = Set.copyOf(peers);
+ this.learners = Set.copyOf(learners);
+ }
+
+ /**
+ * Creates an instance using peers represented as their consistent IDs.
+ */
+ public static PeersAndLearners fromConsistentIds(Set<String> peerNames) {
+ return fromConsistentIds(peerNames, Set.of());
+ }
+
+ /**
+ * Creates an instance using peers and learners represented as their consistent IDs.
Review Comment:
Let's document the index selection logic as it's not too obvious
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -17,75 +17,83 @@
package org.apache.ignite.internal.raft;
-import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.lang.NodeStoppingException;
+import org.jetbrains.annotations.Nullable;
/**
* Raft manager.
*/
public interface RaftManager extends IgniteComponent {
/**
- * Creates a raft group service providing operations on a raft group. If {@code nodes} contains the current node, then raft group starts
- * on the current node.
+ * Creates a raft group service providing operations on a raft group.
*
* @param groupId Raft group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
+ * @param serverPeer Local peer that will host the Raft node. If {@code null} - no nodes will be started, but only the Raft client.
+ * @param configuration Peers and Learners of the Raft group.
* @param lsnrSupplier Raft group listener supplier.
* @return Future representing pending completion of the operation.
* @throws NodeStoppingException If node stopping intention was detected.
*/
CompletableFuture<RaftGroupService> prepareRaftGroup(
ReplicationGroupId groupId,
- Collection<String> peerConsistentIds,
+ @Nullable Peer serverPeer,
+ PeersAndLearners configuration,
Review Comment:
Is it correct to call peers+leaners a configuration? A RAFT configuration is 4 collections of members, not 2 of them. This might cause confusion.
How about `initialMembers`?
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -17,75 +17,83 @@
package org.apache.ignite.internal.raft;
-import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.lang.NodeStoppingException;
+import org.jetbrains.annotations.Nullable;
/**
* Raft manager.
*/
public interface RaftManager extends IgniteComponent {
/**
- * Creates a raft group service providing operations on a raft group. If {@code nodes} contains the current node, then raft group starts
- * on the current node.
+ * Creates a raft group service providing operations on a raft group.
*
* @param groupId Raft group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
+ * @param serverPeer Local peer that will host the Raft node. If {@code null} - no nodes will be started, but only the Raft client.
+ * @param configuration Peers and Learners of the Raft group.
* @param lsnrSupplier Raft group listener supplier.
* @return Future representing pending completion of the operation.
* @throws NodeStoppingException If node stopping intention was detected.
*/
CompletableFuture<RaftGroupService> prepareRaftGroup(
ReplicationGroupId groupId,
- Collection<String> peerConsistentIds,
+ @Nullable Peer serverPeer,
+ PeersAndLearners configuration,
Supplier<RaftGroupListener> lsnrSupplier
) throws NodeStoppingException;
/**
- * Creates a raft group service providing operations on a raft group. If {@code nodeConsistentIds} or {@code learnerConsistentIds}
- * contains the current node, then raft group starts on the current node.
+ * Creates a raft group service providing operations on a raft group.
*
* @param groupId Raft group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
- * @param learnerConsistentIds Consistent IDs of Raft learner nodes.
+ * @param serverPeer Local peer that will host the Raft node. If {@code null} - no nodes will be started, but only the Raft client.
+ * @param configuration Peers and Learners of the Raft group.
* @param lsnrSupplier Raft group listener supplier.
* @param raftGrpEvtsLsnrSupplier Raft group events listener supplier.
* @return Future representing pending completion of the operation.
* @throws NodeStoppingException If node stopping intention was detected.
*/
CompletableFuture<RaftGroupService> prepareRaftGroup(
ReplicationGroupId groupId,
- Collection<String> peerConsistentIds,
- Collection<String> learnerConsistentIds,
+ @Nullable Peer serverPeer,
+ PeersAndLearners configuration,
Supplier<RaftGroupListener> lsnrSupplier,
Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier
) throws NodeStoppingException;
/**
- * Stops a raft group on the current node.
+ * Stops a given local Raft node.
*
* @param groupId Raft group id.
+ * @return {@code true} if the node has been stopped, {@code false} otherwise.
* @throws NodeStoppingException If node stopping intention was detected.
*/
- void stopRaftGroup(ReplicationGroupId groupId) throws NodeStoppingException;
+ boolean stopRaftNode(RaftGroupId groupId) throws NodeStoppingException;
+
+ /**
+ * Stops all local nodes running the given Raft group.
Review Comment:
Should there be an explanation how more than one RAFT node might exist for the same group in the same Ignite instance?
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftGroupId.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.raft;
+
+import java.util.Objects;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.tostring.S;
+
+/**
+ * Class that is used to uniquely identify a locally hosted Raft group.
+ */
+public class RaftGroupId {
Review Comment:
It seems that it actually identifies not a RAFT group, but a node of that group. Should the class be renamed? Maybe `RaftGroupNodeId`/`RaftNodeId`?
##########
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLearnersTest.java:
##########
@@ -273,20 +292,120 @@ public void testLostLearners() throws Exception {
assertThat(services.get(0).thenCompose(RaftGroupService::refreshLeader), willCompleteSuccessfully());
}
+ /**
+ * Tests a situation when a peer and a learner are started on the same node.
+ */
+ @Test
+ void testLearnersOnTheSameNodeAsPeers() throws InterruptedException {
+ RaftNode node = nodes.get(0);
+
+ PeersAndLearners configuration = createConfiguration(List.of(node), List.of(node));
+
+ var peerListener = new TestRaftGroupListener();
+ var learnerListener = new TestRaftGroupListener();
+
+ Peer peer = configuration.peer(node.consistentId());
+ Peer learner = configuration.learner(node.consistentId());
+
+ CompletableFuture<RaftGroupService> peerService = startRaftGroup(node, peer, configuration, peerListener);
+ CompletableFuture<RaftGroupService> learnerService = startRaftGroup(node, learner, configuration, learnerListener);
+
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
Review Comment:
Same as above
##########
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItRaftGroupServiceTest.java:
##########
@@ -228,11 +240,17 @@ CompletableFuture<RaftGroupService> startRaftGroup(List<String> peers, List<Stri
}
void beforeNodeStop() throws Exception {
- IgniteUtils.closeAll(
- raftGroupService == null ? null : () -> loza.stopRaftGroup(RAFT_GROUP_NAME),
- loza::beforeNodeStop,
- clusterService::beforeNodeStop
+ Stream<AutoCloseable> shutdownService = Stream.of(
+ raftGroupService == null
+ ? null
+ : (AutoCloseable) () -> raftGroupService.get(1, TimeUnit.SECONDS).shutdown()
);
+
+ Stream<AutoCloseable> stopRaftGroups = loza.startedGroups().stream().map(id -> () -> loza.stopRaftNode(id));
+
+ Stream<AutoCloseable> beforeNodeStop = Stream.of(loza::beforeNodeStop, clusterService::beforeNodeStop);
+
+ IgniteUtils.closeAll(Stream.of(shutdownService, stopRaftGroups, beforeNodeStop).flatMap(Function.identity()));
Review Comment:
```suggestion
IgniteUtils.closeAll(Stream.concat(shutdownService, stopRaftGroups, beforeNodeStop));
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -248,55 +221,42 @@ public CompletableFuture<RaftGroupService> prepareRaftGroup(
* Internal method to a raft group creation.
*
* @param groupId Raft group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
- * @param learnerConsistentIds Consistent IDs of Raft learners.
+ * @param configuration Peers and Learners of the Raft group.
Review Comment:
`@param localMember` is missing, is this intentional? Also, it should probably be noted in the javadoc what is the difference between passing null and non-null local member.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -126,8 +127,8 @@ private RaftGroupServiceImpl(
RaftMessagesFactory factory,
int timeout,
int rpcTimeout,
- List<Peer> peers,
- List<Peer> learners,
+ Collection<Peer> peers,
Review Comment:
I wonder why is this constructor left with separate peers+learners and not `PeersAndLearners`?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -322,96 +275,91 @@ public void startRaftGroupNode(
/**
* Creates and starts a raft group service providing operations on a raft group.
*
- * @param grpId RAFT group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
- * @param learnerConsistentIds Consistent IDs of Raft learners.
+ * @param groupId RAFT group id.
+ * @param configuration Peers and Learners of the Raft group.
* @return Future that will be completed with an instance of RAFT group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
@Override
public CompletableFuture<RaftGroupService> startRaftGroupService(
- ReplicationGroupId grpId,
- Collection<String> peerConsistentIds,
- Collection<String> learnerConsistentIds
+ ReplicationGroupId groupId,
+ PeersAndLearners configuration
) throws NodeStoppingException {
if (!busyLock.enterBusy()) {
throw new NodeStoppingException();
}
try {
- return startRaftGroupServiceInternal(grpId, idsToPeers(peerConsistentIds), idsToPeers(learnerConsistentIds));
+ return startRaftGroupServiceInternal(groupId, configuration);
} finally {
busyLock.leaveBusy();
}
}
private void startRaftGroupNodeInternal(
- ReplicationGroupId grpId,
- List<Peer> peers,
- List<Peer> learners,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupListener lsnr,
RaftGroupEventsListener raftGrpEvtsLsnr,
RaftGroupOptions groupOptions
) {
- assert !peers.isEmpty();
-
if (LOG.isInfoEnabled()) {
- LOG.info("Start new raft node for group={} with initial peers={}", grpId, peers);
+ LOG.info("Start new raft node for group={} with initial peers={}", groupId, configuration);
}
- boolean started = raftServer.startRaftGroup(grpId, raftGrpEvtsLsnr, lsnr, peers, learners, groupOptions);
+ boolean started = raftServer.startRaftGroup(groupId, configuration, raftGrpEvtsLsnr, lsnr, groupOptions);
if (!started) {
throw new IgniteInternalException(IgniteStringFormatter.format(
"Raft group on the node is already started [raftGrp={}]",
- grpId
+ groupId
));
}
}
- private CompletableFuture<RaftGroupService> startRaftGroupServiceInternal(
- ReplicationGroupId grpId,
- List<Peer> peers,
- List<Peer> learners
- ) {
- assert !peers.isEmpty();
-
+ private CompletableFuture<RaftGroupService> startRaftGroupServiceInternal(ReplicationGroupId grpId, PeersAndLearners configuration) {
return RaftGroupServiceImpl.start(
grpId,
clusterNetSvc,
FACTORY,
RETRY_TIMEOUT,
RPC_TIMEOUT,
- peers,
- learners,
+ configuration,
true,
DELAY,
executor
);
}
- private static List<Peer> idsToPeers(Collection<String> nodes) {
- return nodes.stream().map(Peer::new).collect(Collectors.toUnmodifiableList());
+ @Override
+ public boolean stopRaftNode(RaftGroupId groupId) throws NodeStoppingException {
+ if (!busyLock.enterBusy()) {
+ throw new NodeStoppingException();
+ }
+
+ try {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Stop raft group={}", groupId);
Review Comment:
'Stop raft node'?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/ReadOnlyServiceImpl.java:
##########
@@ -36,9 +36,9 @@
import org.apache.ignite.raft.jraft.ReadOnlyService;
import org.apache.ignite.raft.jraft.Status;
import org.apache.ignite.raft.jraft.closure.ReadIndexClosure;
-import org.apache.ignite.raft.jraft.disruptor.GroupAware;
+import org.apache.ignite.raft.jraft.disruptor.NodeIdAware;
import org.apache.ignite.raft.jraft.disruptor.StripedDisruptor;
-import org.apache.ignite.raft.jraft.entity.ReadIndexState;
+import org.apache.ignite.raft.jraft.entity.NodeId;import org.apache.ignite.raft.jraft.entity.ReadIndexState;
Review Comment:
A few imports in a single line
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java:
##########
@@ -38,8 +38,7 @@
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.raft.JraftGroupEventsListener;import org.apache.ignite.internal.thread.NamedThreadFactory;
-import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.JraftGroupEventsListener;import org.apache.ignite.internal.raft.RaftGroupId;import org.apache.ignite.internal.thread.NamedThreadFactory;
Review Comment:
A few imports in one line? Was IDEA playful? :)
##########
modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ItMetaStorageRaftGroupTest.java:
##########
@@ -323,18 +326,25 @@ private List<Pair<RaftServer, RaftGroupService>> prepareJraftMetaStorages(Atomic
metaStorageRaftSrv3.start();
- metaStorageRaftSrv1.startRaftGroup(INSTANCE, new MetaStorageListener(mockStorage), peers, defaults());
+ var raftGroupId1 = new RaftGroupId(INSTANCE, configuration.peer(localMemberName(cluster.get(0))));
Review Comment:
When reading this code, it's impossible to guess what this `INSTANCE` is. It should probably never be imported statically. Can it be qualified, at least in the classes you touched?
##########
modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterInitializer.java:
##########
@@ -100,8 +101,8 @@ public CompletableFuture<Void> initCluster(
LOG.info("Resolved CMG nodes[nodes={}]", cmgNodes);
CmgInitMessage initMessage = msgFactory.cmgInitMessage()
- .metaStorageNodes(metaStorageNodeNames)
- .cmgNodes(cmgNodeNames)
+ .metaStorageNodes(Set.copyOf(metaStorageNodeNames))
Review Comment:
These sets are already unmodifiable. Why do we need to copy them?
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftGroupId.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.raft;
+
+import java.util.Objects;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.tostring.S;
+
+/**
+ * Class that is used to uniquely identify a locally hosted Raft group.
Review Comment:
A nitpick: actually, instances of the class are used to identify..., not the class itself. It looks like 3 first words of this javadoc could be omitted.
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/PeersAndLearners.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.raft;
+
+import static java.util.stream.Collectors.toUnmodifiableSet;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import org.apache.ignite.internal.tostring.IgniteToStringInclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Class containing peers and learners of a Raft Group.
+ */
+public class PeersAndLearners {
+ @IgniteToStringInclude
+ private final Set<Peer> peers;
+
+ @IgniteToStringInclude
+ private final Set<Peer> learners;
+
+ private PeersAndLearners(Collection<Peer> peers, Collection<Peer> learners) {
+ this.peers = Set.copyOf(peers);
+ this.learners = Set.copyOf(learners);
+ }
+
+ /**
+ * Creates an instance using peers represented as their consistent IDs.
+ */
+ public static PeersAndLearners fromConsistentIds(Set<String> peerNames) {
+ return fromConsistentIds(peerNames, Set.of());
+ }
+
+ /**
+ * Creates an instance using peers and learners represented as their consistent IDs.
+ */
+ public static PeersAndLearners fromConsistentIds(Set<String> peerNames, Set<String> learnerNames) {
+ Set<Peer> peers = peerNames.stream().map(Peer::new).collect(toUnmodifiableSet());
+
+ Set<Peer> learners = learnerNames.stream()
+ .map(name -> {
+ int idx = peerNames.contains(name) ? 1 : 0;
+
+ return new Peer(name, idx);
+ })
+ .collect(toUnmodifiableSet());
+
+ return new PeersAndLearners(peers, learners);
+ }
+
+ /**
+ * Creates an instance using peers and learners represented as {@link Peer}s.
+ */
+ public static PeersAndLearners fromPeers(Collection<Peer> peers, Collection<Peer> learners) {
+ assert Collections.disjoint(peers, learners);
+
+ return new PeersAndLearners(peers, learners);
+ }
+
+ /**
+ * Returns the set of peers.
+ */
+ public Set<Peer> peers() {
+ return peers;
+ }
+
+ /**
+ * Returns the set of learners.
+ */
+ public Set<Peer> learners() {
+ return learners;
+ }
+
+ /**
+ * Returns a peer with the given consistent ID or {@code null} if it is not present in the configuration.
+ */
+ public @Nullable Peer peer(String consistentId) {
+ return peers.stream().filter(p -> p.consistentId().equals(consistentId)).findAny().orElse(null);
+ }
+
+ /**
+ * Returns a learner with the given consistent ID or {@code null} if it is not present in the configuration.
+ */
+ public @Nullable Peer learner(String consistentId) {
+ return learners.stream().filter(p -> p.consistentId().equals(consistentId)).findAny().orElse(null);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ PeersAndLearners that = (PeersAndLearners) o;
+
+ if (!peers.equals(that.peers)) {
Review Comment:
Again, about priorities: two peers with same consistentId+index, but different priorities, will be different. Is this ok?
##########
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLearnersTest.java:
##########
@@ -273,20 +292,120 @@ public void testLostLearners() throws Exception {
assertThat(services.get(0).thenCompose(RaftGroupService::refreshLeader), willCompleteSuccessfully());
}
+ /**
+ * Tests a situation when a peer and a learner are started on the same node.
+ */
+ @Test
+ void testLearnersOnTheSameNodeAsPeers() throws InterruptedException {
+ RaftNode node = nodes.get(0);
+
+ PeersAndLearners configuration = createConfiguration(List.of(node), List.of(node));
+
+ var peerListener = new TestRaftGroupListener();
+ var learnerListener = new TestRaftGroupListener();
+
+ Peer peer = configuration.peer(node.consistentId());
+ Peer learner = configuration.learner(node.consistentId());
+
+ CompletableFuture<RaftGroupService> peerService = startRaftGroup(node, peer, configuration, peerListener);
+ CompletableFuture<RaftGroupService> learnerService = startRaftGroup(node, learner, configuration, learnerListener);
+
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
Review Comment:
This seems excessive. We already asserted that the leader is `peer`, so it cannot be `learner`
##########
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLearnersTest.java:
##########
@@ -273,20 +292,120 @@ public void testLostLearners() throws Exception {
assertThat(services.get(0).thenCompose(RaftGroupService::refreshLeader), willCompleteSuccessfully());
}
+ /**
+ * Tests a situation when a peer and a learner are started on the same node.
+ */
+ @Test
+ void testLearnersOnTheSameNodeAsPeers() throws InterruptedException {
+ RaftNode node = nodes.get(0);
+
+ PeersAndLearners configuration = createConfiguration(List.of(node), List.of(node));
+
+ var peerListener = new TestRaftGroupListener();
+ var learnerListener = new TestRaftGroupListener();
+
+ Peer peer = configuration.peer(node.consistentId());
+ Peer learner = configuration.learner(node.consistentId());
+
+ CompletableFuture<RaftGroupService> peerService = startRaftGroup(node, peer, configuration, peerListener);
+ CompletableFuture<RaftGroupService> learnerService = startRaftGroup(node, learner, configuration, learnerListener);
+
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
+
+ // Test writing data.
+ CompletableFuture<?> writeFuture = peerService
+ .thenCompose(s -> s.run(createWriteCommand("foo")).thenApply(v -> s))
+ .thenCompose(s -> s.run(createWriteCommand("bar")));
+
+ assertThat(writeFuture, willCompleteSuccessfully());
+
+ for (TestRaftGroupListener listener : Arrays.asList(peerListener, learnerListener)) {
+ assertThat(listener.storage.poll(1, TimeUnit.SECONDS), is("foo"));
+ assertThat(listener.storage.poll(1, TimeUnit.SECONDS), is("bar"));
+ }
+ }
+
+ /**
+ * Tests adding a new learner to a node that already runs a Raft node.
+ */
+ @Test
+ void testAddLearnerToSameNodeAsPeer() throws InterruptedException {
+ List<RaftNode> followers = nodes.subList(0, 2);
+ RaftNode learner = nodes.get(0);
+
+ PeersAndLearners configuration = createConfiguration(followers, List.of(learner));
+
+ CompletableFuture<?>[] followerServices = followers.stream()
+ .map(node -> startRaftGroup(node, configuration.peer(node.consistentId()), configuration, new TestRaftGroupListener()))
+ .toArray(CompletableFuture[]::new);
+
+ assertThat(CompletableFuture.allOf(followerServices), willCompleteSuccessfully());
+
+ var learnerListener = new TestRaftGroupListener();
+
+ CompletableFuture<RaftGroupService> learnerService = startRaftGroup(
+ learner, configuration.learner(learner.consistentId()), configuration, learnerListener
+ );
+
+ CompletableFuture<?> writeFuture = learnerService
+ .thenCompose(s -> s.run(createWriteCommand("foo")).thenApply(v -> s))
+ .thenCompose(s -> s.run(createWriteCommand("bar")));
+
+ assertThat(writeFuture, willCompleteSuccessfully());
+ assertThat(learnerListener.storage.poll(1, TimeUnit.SECONDS), is("foo"));
+ assertThat(learnerListener.storage.poll(1, TimeUnit.SECONDS), is("bar"));
+
+ // Create a new learner on the second node.
+ RaftNode newLearner = nodes.get(1);
+
+ PeersAndLearners newConfiguration = createConfiguration(followers, List.of(learner, newLearner));
+
+ CompletableFuture<Void> changePeersFuture = learnerService.thenCompose(s -> s.refreshAndGetLeaderWithTerm()
+ .thenCompose(leaderWithTerm -> s.changePeersAsync(newConfiguration, leaderWithTerm.term())
+ ));
+
+ assertThat(changePeersFuture, willCompleteSuccessfully());
+
+ var newLearnerListener = new TestRaftGroupListener();
+
+ CompletableFuture<RaftGroupService> newLearnerService = startRaftGroup(
+ newLearner, newConfiguration.learner(newLearner.consistentId()), newConfiguration, newLearnerListener
+ );
+
+ assertThat(newLearnerService, willCompleteSuccessfully());
+ assertThat(newLearnerListener.storage.poll(10, TimeUnit.SECONDS), is("foo"));
+ assertThat(newLearnerListener.storage.poll(10, TimeUnit.SECONDS), is("bar"));
+ }
+
+ private PeersAndLearners createConfiguration(Collection<RaftNode> peers, Collection<RaftNode> learners) {
+ return PeersAndLearners.fromConsistentIds(
+ peers.stream().map(RaftNode::consistentId).collect(toSet()),
+ learners.stream().map(RaftNode::consistentId).collect(toSet())
+ );
+ }
+
+ private List<Peer> getPeers(PeersAndLearners configuration, Collection<RaftNode> peers, Collection<RaftNode> learners) {
Review Comment:
The method concats both peers and learners, so `getPeers` does not seem to be a perfect name. How about `getMembers`?
##########
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLearnersTest.java:
##########
@@ -273,20 +292,120 @@ public void testLostLearners() throws Exception {
assertThat(services.get(0).thenCompose(RaftGroupService::refreshLeader), willCompleteSuccessfully());
}
+ /**
+ * Tests a situation when a peer and a learner are started on the same node.
+ */
+ @Test
+ void testLearnersOnTheSameNodeAsPeers() throws InterruptedException {
+ RaftNode node = nodes.get(0);
+
+ PeersAndLearners configuration = createConfiguration(List.of(node), List.of(node));
+
+ var peerListener = new TestRaftGroupListener();
+ var learnerListener = new TestRaftGroupListener();
+
+ Peer peer = configuration.peer(node.consistentId());
+ Peer learner = configuration.learner(node.consistentId());
+
+ CompletableFuture<RaftGroupService> peerService = startRaftGroup(node, peer, configuration, peerListener);
+ CompletableFuture<RaftGroupService> learnerService = startRaftGroup(node, learner, configuration, learnerListener);
+
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(peerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(peer));
+ assertThat(learnerService.thenApply(RaftGroupService::leader), willBe(not(learner)));
+
+ // Test writing data.
+ CompletableFuture<?> writeFuture = peerService
+ .thenCompose(s -> s.run(createWriteCommand("foo")).thenApply(v -> s))
+ .thenCompose(s -> s.run(createWriteCommand("bar")));
+
+ assertThat(writeFuture, willCompleteSuccessfully());
+
+ for (TestRaftGroupListener listener : Arrays.asList(peerListener, learnerListener)) {
+ assertThat(listener.storage.poll(1, TimeUnit.SECONDS), is("foo"));
+ assertThat(listener.storage.poll(1, TimeUnit.SECONDS), is("bar"));
+ }
+ }
+
+ /**
+ * Tests adding a new learner to a node that already runs a Raft node.
+ */
+ @Test
+ void testAddLearnerToSameNodeAsPeer() throws InterruptedException {
Review Comment:
'add learner to same node as peer' might be understood as 'we have a learner on node A, and we add a peer to the same node A', but the test does something different: it seems to add a learner to a group that already has a learner (on another node) and a peer on the node where learner is added.
From the javadoc I get a third idea about what's going on :) A node already runs a Raft node? There is already a test above that makes sure that a learner can be added to node A when the node A runs a peer...
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -171,52 +168,30 @@ public void stop() throws Exception {
@Override
public CompletableFuture<RaftGroupService> prepareRaftGroup(
ReplicationGroupId groupId,
- Collection<String> nodeConsistentIds,
+ @Nullable Peer serverPeer,
+ PeersAndLearners configuration,
Supplier<RaftGroupListener> lsnrSupplier
) throws NodeStoppingException {
- return prepareRaftGroup(groupId, nodeConsistentIds, lsnrSupplier, RaftGroupOptions.defaults());
+ return prepareRaftGroup(groupId, serverPeer, configuration, lsnrSupplier, () -> noopLsnr, RaftGroupOptions.defaults());
}
@Override
public CompletableFuture<RaftGroupService> prepareRaftGroup(
ReplicationGroupId groupId,
- Collection<String> nodeConsistentIds,
- Collection<String> learnerConsistentIds,
+ @Nullable Peer serverPeer,
+ PeersAndLearners configuration,
Supplier<RaftGroupListener> lsnrSupplier,
Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier
) throws NodeStoppingException {
- return prepareRaftGroup(
- groupId, nodeConsistentIds, learnerConsistentIds, lsnrSupplier, raftGrpEvtsLsnrSupplier, RaftGroupOptions.defaults()
- );
- }
-
- /**
- * Creates a raft group service providing operations on a raft group. If {@code nodes} contains the current node, then raft group starts
- * on the current node.
- *
- * @param groupId Raft group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
- * @param lsnrSupplier Raft group listener supplier.
- * @param groupOptions Options to apply to the group.
- * @return Future representing pending completion of the operation.
- * @throws NodeStoppingException If node stopping intention was detected.
- */
- public CompletableFuture<RaftGroupService> prepareRaftGroup(
- ReplicationGroupId groupId,
- Collection<String> peerConsistentIds,
- Supplier<RaftGroupListener> lsnrSupplier,
- RaftGroupOptions groupOptions
- ) throws NodeStoppingException {
- return prepareRaftGroup(groupId, peerConsistentIds, List.of(), lsnrSupplier, () -> noopLsnr, groupOptions);
+ return prepareRaftGroup(groupId, serverPeer, configuration, lsnrSupplier, raftGrpEvtsLsnrSupplier, RaftGroupOptions.defaults());
}
/**
* Creates a raft group service providing operations on a raft group. If {@code peerConsistentIds} or {@code learnerConsistentIds}
Review Comment:
Part about 'contains' should probably be changed
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java:
##########
@@ -43,59 +44,65 @@ public interface RaftServer extends IgniteComponent {
* Starts a raft group bound to this cluster node.
*
* @param groupId Group id.
- * @param lsnr The listener.
- * @param peers Peers configuration.
+ * @param configuration Raft peers configuration.
+ * @param lsnr Listener for state machine events.
* @param groupOptions Options to apply to the group.
* @return {@code True} if a group was successfully started, {@code False} when the group with given name is already exists.
*/
boolean startRaftGroup(
- ReplicationGroupId groupId,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupListener lsnr,
- List<Peer> peers,
RaftGroupOptions groupOptions
);
/**
* Starts a raft group bound to this cluster node.
*
* @param groupId Group id.
+ * @param configuration Raft peers configuration.
* @param evLsnr Listener for group membership and other events.
* @param lsnr Listener for state machine events.
- * @param peers Peers configuration.
- * @param learners Learners configuration.
* @param groupOptions Options to apply to the group.
* @return {@code True} if a group was successfully started, {@code False} when the group with given name is already exists.
*/
boolean startRaftGroup(
- ReplicationGroupId groupId,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupEventsListener evLsnr,
RaftGroupListener lsnr,
- List<Peer> peers,
- List<Peer> learners,
RaftGroupOptions groupOptions
);
/**
- * Synchronously stops a raft group if any.
+ * Stops a given local Raft node if it exists.
*
* @param groupId Group id.
- * @return {@code True} if a group was successfully stopped.
+ * @return {@code true} if the node has been stopped, {@code false} otherwise.
+ */
+ boolean stopRaftNode(RaftGroupId groupId);
+
+ /**
+ * Stops all local nodes running the given Raft group.
+ *
+ * @param replicationGroupId Raft group name.
+ * @return {@code true} if at least one node has been stopped, {@code false} otherwise.
*/
- boolean stopRaftGroup(ReplicationGroupId groupId);
+ boolean stopRaftNodes(ReplicationGroupId replicationGroupId);
/**
- * Returns a local peer.
+ * Returns local nodes running the given Raft group.
*
* @param groupId Group id.
- * @return Local peer or null if the group is not started.
+ * @return List of peers (can be empty if no local Raft nodes have been started).
*/
- @Nullable Peer localPeer(ReplicationGroupId groupId);
+ List<Peer> localPeers(ReplicationGroupId groupId);
/**
* Returns a set of started partition groups.
*
* @return Started groups.
*/
@TestOnly
- Set<ReplicationGroupId> startedGroups();
+ Set<RaftGroupId> startedGroups();
Review Comment:
Started nodes?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java:
##########
@@ -322,96 +275,91 @@ public void startRaftGroupNode(
/**
* Creates and starts a raft group service providing operations on a raft group.
*
- * @param grpId RAFT group id.
- * @param peerConsistentIds Consistent IDs of Raft peers.
- * @param learnerConsistentIds Consistent IDs of Raft learners.
+ * @param groupId RAFT group id.
+ * @param configuration Peers and Learners of the Raft group.
* @return Future that will be completed with an instance of RAFT group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
@Override
public CompletableFuture<RaftGroupService> startRaftGroupService(
- ReplicationGroupId grpId,
- Collection<String> peerConsistentIds,
- Collection<String> learnerConsistentIds
+ ReplicationGroupId groupId,
+ PeersAndLearners configuration
) throws NodeStoppingException {
if (!busyLock.enterBusy()) {
throw new NodeStoppingException();
}
try {
- return startRaftGroupServiceInternal(grpId, idsToPeers(peerConsistentIds), idsToPeers(learnerConsistentIds));
+ return startRaftGroupServiceInternal(groupId, configuration);
} finally {
busyLock.leaveBusy();
}
}
private void startRaftGroupNodeInternal(
- ReplicationGroupId grpId,
- List<Peer> peers,
- List<Peer> learners,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupListener lsnr,
RaftGroupEventsListener raftGrpEvtsLsnr,
RaftGroupOptions groupOptions
) {
- assert !peers.isEmpty();
-
if (LOG.isInfoEnabled()) {
- LOG.info("Start new raft node for group={} with initial peers={}", grpId, peers);
+ LOG.info("Start new raft node for group={} with initial peers={}", groupId, configuration);
Review Comment:
It's not just 'peers' now
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -98,7 +98,7 @@ public class JraftServerImpl implements RaftServer {
private IgniteRpcServer rpcServer;
/** Started groups. */
- private final ConcurrentMap<ReplicationGroupId, RaftGroupService> groups = new ConcurrentHashMap<>();
+ private final ConcurrentMap<RaftGroupId, RaftGroupService> groups = new ConcurrentHashMap<>();
Review Comment:
It looks like it now stores nodes, not groups
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/RaftServer.java:
##########
@@ -43,59 +44,65 @@ public interface RaftServer extends IgniteComponent {
* Starts a raft group bound to this cluster node.
*
* @param groupId Group id.
- * @param lsnr The listener.
- * @param peers Peers configuration.
+ * @param configuration Raft peers configuration.
+ * @param lsnr Listener for state machine events.
* @param groupOptions Options to apply to the group.
* @return {@code True} if a group was successfully started, {@code False} when the group with given name is already exists.
*/
boolean startRaftGroup(
- ReplicationGroupId groupId,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupListener lsnr,
- List<Peer> peers,
RaftGroupOptions groupOptions
);
/**
* Starts a raft group bound to this cluster node.
*
* @param groupId Group id.
+ * @param configuration Raft peers configuration.
* @param evLsnr Listener for group membership and other events.
* @param lsnr Listener for state machine events.
- * @param peers Peers configuration.
- * @param learners Learners configuration.
* @param groupOptions Options to apply to the group.
* @return {@code True} if a group was successfully started, {@code False} when the group with given name is already exists.
*/
boolean startRaftGroup(
- ReplicationGroupId groupId,
+ RaftGroupId groupId,
+ PeersAndLearners configuration,
RaftGroupEventsListener evLsnr,
RaftGroupListener lsnr,
- List<Peer> peers,
- List<Peer> learners,
RaftGroupOptions groupOptions
);
/**
- * Synchronously stops a raft group if any.
+ * Stops a given local Raft node if it exists.
Review Comment:
Wasn't that 'synchronously' important here?
##########
modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/raft/ItCmgRaftServiceTest.java:
##########
@@ -178,9 +177,10 @@ void setUp(@WorkDirectory Path workDir, TestInfo testInfo) {
}
@AfterEach
- void tearDown() {
- cluster.parallelStream().forEach(Node::beforeNodeStop);
- cluster.parallelStream().forEach(Node::stop);
+ void tearDown() throws Exception {
+ IgniteUtils.closeAll(cluster.parallelStream().map(node -> node::beforeNodeStop));
Review Comment:
This starts smelling like an abuse. `closeAll()` is used to run a few actions at the best effort basis, it's not about closing anymore. How about adding a method like `runWithBestEffort()` accepting something analogous to `RunnableX` (a runnable with `throws Exception`)?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -474,21 +502,25 @@ private <R extends NetworkMessage> CompletableFuture<R> sendWithRetry(Peer peer,
* Retries a request until success or timeout.
*
* @param peer Target peer.
Review Comment:
This seems to be the target peer used on the *first* attempt; but on subsequent attempts, other peers will be used. I think this should be reflected in the javadoc of this method and/or in the naming of the parameter (like `initialPeer`).
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2113,4 +2094,19 @@ private static TxStateStorage getOrCreateTxStateStorage(TxStateTableStorage txSt
private CompletableFuture<TxStateStorage> getOrCreateTxStateStorageAsync(TxStateTableStorage txStateTableStorage, int partId) {
return CompletableFuture.supplyAsync(() -> getOrCreateTxStateStorage(txStateTableStorage, partId), ioExecutor);
}
+
+ private static PeersAndLearners parseRaftConfiguration(Collection<Assignment> assignments) {
Review Comment:
Why is it called 'parse'? It does not seem to make any string parsing. Is it parsing of some other kind?
How about `assignmentsToConfiguration`?
--
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