You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by as...@apache.org on 2021/02/27 11:23:46 UTC
[ignite-3] branch ignite-14149 updated (d70fe7d -> c4871ce)
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a change to branch ignite-14149
in repository https://gitbox.apache.org/repos/asf/ignite-3.git.
from d70fe7d IGNITE-14149 Services.
new 092bc39 IGNITE-14149 Cleanup.
new df9226b IGNITE-14149 wip.
new ac96cae IGNITE-14149 wip.
new c4871ce IGNITE-14149 wip.
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/ignite/raft/PeerId.java | 4 +
.../main/java/org/apache/ignite/raft/State.java | 265 +--------------------
...CommonMessages.java => RaftClientMessages.java} | 8 +-
.../client/message/AddLearnersRequestImpl.java | 7 +-
.../raft/client/message/AddPeerRequestImpl.java | 6 +-
.../raft/client/message/AddPeerResponseImpl.java | 6 +-
.../raft/client/message/ChangePeerRequestImpl.java | 6 +-
.../client/message/ChangePeersResponseImpl.java | 6 +-
.../message/ClientMessageBuilderFactory.java | 44 ++--
.../raft/client/message/GetLeaderRequestImpl.java | 6 +-
.../raft/client/message/GetLeaderResponseImpl.java | 6 +-
.../raft/client/message/GetPeersRequestImpl.java | 6 +-
.../raft/client/message/GetPeersResponseImpl.java | 6 +-
.../client/message/LearnersOpResponseImpl.java | 6 +-
.../raft/client/message/PingRequestImpl.java | 6 +-
.../RaftClientCommonMessageBuilderFactory.java | 86 -------
.../message/RaftClientMessageBuilderFactory.java | 94 ++++++++
.../client/message/RemoveLearnersRequestImpl.java | 6 +-
.../raft/client/message/RemovePeerRequestImpl.java | 6 +-
.../client/message/RemovePeerResponseImpl.java | 6 +-
.../client/message/ResetLearnersRequestImpl.java | 6 +-
.../raft/client/message/ResetPeerRequestImpl.java | 6 +-
.../raft/client/message/SnapshotRequestImpl.java | 7 +-
.../raft/client/message/StatusResponseImpl.java | 6 +-
.../client/message/TransferLeaderRequestImpl.java | 6 +-
.../raft/client/message/UserRequestImpl.java | 32 +++
.../raft/client/message/UserResponseImpl.java | 21 ++
.../ignite/raft/client/rpc/RaftGroupRpcClient.java | 50 ++--
.../client/rpc/impl/RaftGroupRpcClientImpl.java | 144 +++++++----
.../service/RaftGroupClientRequestListener.java | 2 +-
.../service/RaftGroupClientRequestService.java | 7 +
.../client/service/RaftGroupManagmentService.java | 8 +-
.../impl/RaftGroupClientRequestServiceImpl.java | 20 +-
.../impl/RaftGroupManagementServiceImpl.java | 8 +-
.../java/org/apache/ignite/raft/rpc/NodeImpl.java | 2 +-
.../apache/ignite/raft/rpc/RaftGroupMessage.java | 3 +
.../java/org/apache/ignite/raft/rpc/RpcClient.java | 8 +-
.../ignite/raft/client/RaftGroupRpcClientTest.java | 98 --------
.../raft/client/rpc/RaftGroupRpcClientTest.java | 183 ++++++++++++++
.../service/RaftGroupClientRequestServiceTest.java | 132 ++++++++++
40 files changed, 721 insertions(+), 614 deletions(-)
rename modules/raft-client/src/main/java/org/apache/ignite/raft/client/{RaftClientCommonMessages.java => RaftClientMessages.java} (97%)
delete mode 100644 modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
create mode 100644 modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientMessageBuilderFactory.java
create mode 100644 modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserRequestImpl.java
create mode 100644 modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserResponseImpl.java
delete mode 100644 modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
create mode 100644 modules/raft-client/src/test/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClientTest.java
create mode 100644 modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestServiceTest.java
[ignite-3] 01/04: IGNITE-14149 Cleanup.
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch ignite-14149
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 092bc39b09739a5171b11316cec0c96615dfb13b
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Sat Feb 27 12:32:06 2021 +0300
IGNITE-14149 Cleanup.
---
.../main/java/org/apache/ignite/raft/PeerId.java | 4 +
.../main/java/org/apache/ignite/raft/State.java | 265 +--------------------
.../raft/client/RaftClientCommonMessages.java | 2 +-
.../RaftClientCommonMessageBuilderFactory.java | 2 -
.../ignite/raft/client/rpc/RaftGroupRpcClient.java | 6 +-
.../client/rpc/impl/RaftGroupRpcClientImpl.java | 117 +++++----
.../service/RaftGroupClientRequestListener.java | 2 +-
.../service/RaftGroupClientRequestService.java | 7 +
.../impl/RaftGroupManagementServiceImpl.java | 8 +-
.../java/org/apache/ignite/raft/rpc/NodeImpl.java | 2 +-
.../apache/ignite/raft/rpc/RaftGroupMessage.java | 3 +
.../java/org/apache/ignite/raft/rpc/RpcClient.java | 8 +-
.../ignite/raft/client/RaftGroupRpcClientTest.java | 116 ++++++++-
13 files changed, 212 insertions(+), 330 deletions(-)
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/PeerId.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/PeerId.java
index 56d96be..7b5fec0 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/PeerId.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/PeerId.java
@@ -75,4 +75,8 @@ public class PeerId implements Serializable {
result = 31 * result + priority;
return result;
}
+
+ @Override public String toString() {
+ return node.id() + ":" + priority;
+ }
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/State.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/State.java
index 5f5fb15..5001d74 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/State.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/State.java
@@ -16,271 +16,16 @@
*/
package org.apache.ignite.raft;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
-import java.util.Set;
+import org.jetbrains.annotations.Nullable;
/**
* Raft group state.
*/
-public class State implements Iterable<PeerId> {
- /** Mark a leaner peer */
- private static final String LEARNER_POSTFIX = "/learner";
+public interface State {
+ @Nullable PeerId leader();
- private volatile PeerId leader;
+ @Nullable List<PeerId> peers();
- private List<PeerId> peers = new ArrayList<>();
-
- // use LinkedHashSet to keep insertion order.
- private List<PeerId> learners = new ArrayList<>();
-
- public State() {
- super();
- }
-
- /**
- * Construct a configuration instance with peers.
- *
- * @param conf configuration
- */
- public State(final Iterable<PeerId> conf) {
- this(conf, null);
- }
-
- /**
- * Construct a configuration from another conf.
- *
- * @param conf configuration
- */
- public State(final State conf) {
- this(conf.getPeers(), conf.getLearners());
- }
-
- /**
- * Construct a Configuration instance with peers and learners.
- *
- * @param conf peers configuration
- * @param learners learners
- * @since 1.3.0
- */
- public State(final Iterable<PeerId> conf, final Iterable<PeerId> learners) {
- for (final PeerId peer : conf) {
- this.peers.add(new PeerId(peer));
- }
- addLearners(learners);
- }
-
- public PeerId getLeader() {
- return leader;
- }
-
- public void setLeader(PeerId leader) {
- this.leader = leader;
- }
-
- public void setLearners(final List<PeerId> learners) {
- this.learners = learners;
- }
-
- /**
- * Add a learner peer.
- *
- * @param learner learner to add
- * @return true when add successfully.
- */
- public boolean addLearner(final PeerId learner) {
- return this.learners.add(learner);
- }
-
- /**
- * Add learners in batch, returns the added count.
- *
- * @param learners learners to add
- * @return the total added count
- */
- public int addLearners(final Iterable<PeerId> learners) {
- int ret = 0;
- if (learners != null) {
- for (final PeerId peer : learners) {
- if (this.learners.add(new PeerId(peer))) {
- ret++;
- }
- }
- }
- return ret;
- }
-
- /**
- * Remove a learner peer.
- *
- * @param learner learner to remove
- * @return true when remove successfully.
- */
- public boolean removeLearner(final PeerId learner) {
- return this.learners.remove(learner);
- }
-
- /**
- * Retrieve the learners set.
- *
- * @return learners
- */
- public List<PeerId> getLearners() {
- return this.learners;
- }
-
- /**
- * Retrieve the learners set copy.
- *
- * @return learners
- */
- public List<PeerId> listLearners() {
- return new ArrayList<>(this.learners);
- }
-
- /**
- * Returns true when the configuration is valid.
- *
- * @return true if the configuration is valid.
- */
- public boolean isValid() {
- final Set<PeerId> intersection = new HashSet<>(this.peers);
- intersection.retainAll(this.learners);
- return !this.peers.isEmpty() && intersection.isEmpty();
- }
-
- public void reset() {
- this.peers.clear();
- this.learners.clear();
- }
-
- public boolean isEmpty() {
- return this.peers.isEmpty();
- }
-
- /**
- * Returns the peers total number.
- *
- * @return total num of peers
- */
- public int size() {
- return this.peers.size();
- }
-
- @Override
- public Iterator<PeerId> iterator() {
- return this.peers.iterator();
- }
-
- public Set<PeerId> getPeerSet() {
- return new HashSet<>(this.peers);
- }
-
- public List<PeerId> listPeers() {
- return new ArrayList<>(this.peers);
- }
-
- public List<PeerId> getPeers() {
- return this.peers;
- }
-
- public void setPeers(final List<PeerId> peers) {
- this.peers.clear();
- for (final PeerId peer : peers) {
- this.peers.add(new PeerId(peer));
- }
- }
-
- public void appendPeers(final Collection<PeerId> set) {
- this.peers.addAll(set);
- }
-
- public boolean addPeer(final PeerId peer) {
- return this.peers.add(peer);
- }
-
- public boolean removePeer(final PeerId peer) {
- return this.peers.remove(peer);
- }
-
- public boolean contains(final PeerId peer) {
- return this.peers.contains(peer);
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + ((this.learners == null) ? 0 : this.learners.hashCode());
- result = prime * result + ((this.peers == null) ? 0 : this.peers.hashCode());
- return result;
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
- State other = (State) obj;
- if (this.learners == null) {
- if (other.learners != null) {
- return false;
- }
- } else if (!this.learners.equals(other.learners)) {
- return false;
- }
- if (this.peers == null) {
- return other.peers == null;
- } else {
- return this.peers.equals(other.peers);
- }
- }
-
- @Override
- public String toString() {
- final StringBuilder sb = new StringBuilder();
- final List<PeerId> peers = listPeers();
- int i = 0;
- int size = peers.size();
- for (final PeerId peer : peers) {
- sb.append(peer);
- if (i < size - 1 || !this.learners.isEmpty()) {
- sb.append(",");
- }
- i++;
- }
-
- size = this.learners.size();
- i = 0;
- for (final PeerId peer : this.learners) {
- sb.append(peer).append(LEARNER_POSTFIX);
- if (i < size - 1) {
- sb.append(",");
- }
- i++;
- }
-
- return sb.toString();
- }
-
- /**
- * Get the difference between |*this| and |rhs|
- * |included| would be assigned to |*this| - |rhs|
- * |excluded| would be assigned to |rhs| - |*this|
- */
- public void diff(final State rhs, final State included, final State excluded) {
- included.peers = new ArrayList<>(this.peers);
- included.peers.removeAll(rhs.peers);
- excluded.peers = new ArrayList<>(rhs.peers);
- excluded.peers.removeAll(this.peers);
- }
+ @Nullable List<PeerId> learners();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java
index 30b1cb5..5a40dc6 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java
@@ -257,7 +257,7 @@ public final class RaftClientCommonMessages {
}
}
- public interface UserRequest<T> extends Message {
+ public interface UserRequest<T> extends RaftGroupMessage {
T request();
public interface Builder<T> {
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
index 45f070a..440cee2 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
@@ -6,8 +6,6 @@ import org.apache.ignite.raft.client.RaftClientCommonMessages;
* Raft client message builders factory.
*/
public class RaftClientCommonMessageBuilderFactory implements ClientMessageBuilderFactory {
- public static RaftClientCommonMessageBuilderFactory DEFAULT = new RaftClientCommonMessageBuilderFactory();
-
@Override public RaftClientCommonMessages.PingRequest.Builder createPingRequest() {
return new PingRequestImpl();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
index 986fd71..5f657eb 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
@@ -46,12 +46,12 @@ import static org.apache.ignite.raft.client.RaftClientCommonMessages.TransferLea
public interface RaftGroupRpcClient {
/**
* @param groupId Group id.
- * @return Current group state or null if state is not yet initalized.
+ * @return Group state snapshot.
*/
- @Nullable State state(String groupId);
+ State state(String groupId);
/**
- * Refreshes a state of initialized group.
+ * Refreshes a group leader.
* @param groupId Group id.
* @return A future.
*/
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
index d7dc4c8..f0ceda6 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
@@ -1,6 +1,7 @@
package org.apache.ignite.raft.client.rpc.impl;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -8,20 +9,25 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
-import java.util.function.Function;
import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.RaftException;
import org.apache.ignite.raft.State;
import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientCommonMessages.GetLeaderResponse;
import org.apache.ignite.raft.client.RaftClientCommonMessages.StatusResponse;
+import org.apache.ignite.raft.client.message.ClientMessageBuilderFactory;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
-import org.apache.ignite.raft.client.message.RaftClientCommonMessageBuilderFactory;
import org.apache.ignite.raft.rpc.InvokeCallback;
import org.apache.ignite.raft.rpc.Message;
import org.apache.ignite.raft.rpc.Node;
import org.apache.ignite.raft.rpc.RaftGroupMessage;
import org.apache.ignite.raft.rpc.RpcClient;
+import org.jetbrains.annotations.Nullable;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.raft.client.message.RaftClientCommonMessageBuilderFactory.DEFAULT;
public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
private final ExecutorService executor;
@@ -31,7 +37,10 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
/** Where to ask for initial configuration. */
private final Set<Node> initialCfgNodes;
- private Map<String, State> states = new ConcurrentHashMap<>();
+ /** */
+ private final ClientMessageBuilderFactory factory;
+
+ private Map<String, StateImpl> states = new ConcurrentHashMap<>();
/**
* Accepts dependencies in constructor.
@@ -39,17 +48,22 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
* @param defaultTimeout
* @param initialCfgNode Initial configuration nodes.
*/
- public RaftGroupRpcClientImpl(RpcClient rpcClient, int defaultTimeout, Set<Node> initialCfgNodes) {
+ public RaftGroupRpcClientImpl(RpcClient rpcClient, ClientMessageBuilderFactory factory, int defaultTimeout, Set<Node> initialCfgNodes) {
this.defaultTimeout = defaultTimeout;
this.rpcClient = rpcClient;
+ this.factory = factory;
this.initialCfgNodes = new HashSet<>(initialCfgNodes);
executor = Executors.newWorkStealingPool();
}
@Override public State state(String groupId) {
- State newState = new State();
+ return getState(groupId);
+ }
+
+ private StateImpl getState(String groupId) {
+ StateImpl newState = new StateImpl();
- State state = states.putIfAbsent(groupId, newState);
+ StateImpl state = states.putIfAbsent(groupId, newState);
if (state == null)
state = newState;
@@ -58,28 +72,16 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
}
@Override public CompletableFuture<PeerId> refreshLeader(String groupId) {
- State state = state(groupId);
+ StateImpl state = getState(groupId);
- if (state.getLeader() == null) {
- synchronized (state) {
- PeerId leader = state.getLeader();
+ return refreshLeader(initialCfgNodes.iterator().next(), groupId).
+ thenApply(resp -> {
+ PeerId leaderId = resp.getLeaderId();
- if (leader == null) {
- return refreshLeader(initialCfgNodes.iterator().next(), groupId).
- thenApply(new Function<RaftClientCommonMessages.GetLeaderResponse, PeerId>() {
- @Override public PeerId apply(RaftClientCommonMessages.GetLeaderResponse getLeaderResponse) {
- PeerId leaderId = getLeaderResponse.getLeaderId();
+ state.leader = leaderId;
- state.setLeader(leaderId);
-
- return leaderId;
- }
- });
- }
- }
- }
-
- return CompletableFuture.completedFuture(state.getLeader());
+ return leaderId;
+ });
}
@Override public CompletableFuture<RaftClientCommonMessages.AddPeerResponse> addPeer(RaftClientCommonMessages.AddPeerRequest request) {
@@ -118,21 +120,34 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return null;
}
- private CompletableFuture<RaftClientCommonMessages.GetLeaderResponse> refreshLeader(Node node, String groupId) {
- RaftClientCommonMessages.GetLeaderRequest req = RaftClientCommonMessageBuilderFactory.DEFAULT.createGetLeaderRequest().setGroupId(groupId).build();
+ private CompletableFuture<GetLeaderResponse> refreshLeader(Node node, String groupId) {
+ StateImpl state = getState(groupId);
- CompletableFuture<RaftClientCommonMessages.GetLeaderResponse> fut = new CompletableFuture<>();
+ while(true) {
+ CompletableFuture<GetLeaderResponse> fut = state.updateFutRef.get();
- rpcClient.invokeAsync(node, req, new InvokeCallback<RaftClientCommonMessages.GetLeaderResponse>() {
- @Override public void complete(RaftClientCommonMessages.GetLeaderResponse response, Throwable err) {
- if (err != null)
- fut.completeExceptionally(err);
- else
- fut.complete(response);
- }
- }, executor, defaultTimeout);
+ if (fut != null)
+ return fut;
- return fut;
+ if (state.updateFutRef.compareAndSet(null, (fut = new CompletableFuture<>()))) {
+ RaftClientCommonMessages.GetLeaderRequest req = DEFAULT.createGetLeaderRequest().setGroupId(groupId).build();
+
+ CompletableFuture<GetLeaderResponse> finalFut = fut;
+
+ rpcClient.invokeAsync(node, req, new InvokeCallback<GetLeaderResponse>() {
+ @Override public void complete(GetLeaderResponse response, Throwable err) {
+ if (err != null)
+ finalFut.completeExceptionally(err);
+ else
+ finalFut.complete(response);
+
+ state.updateFutRef.set(null);
+ }
+ }, executor, defaultTimeout);
+
+ return fut;
+ }
+ }
}
@Override public <R extends Message> CompletableFuture<R> sendCustom(RaftGroupMessage request) {
@@ -142,13 +157,13 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
fut.orTimeout(defaultTimeout, TimeUnit.MILLISECONDS);
- CompletableFuture<PeerId> fut0 = refreshLeader(request.getGroupId());
+ CompletableFuture<PeerId> fut0 = state.leader() == null ?
+ refreshLeader(request.getGroupId()) : completedFuture(state.leader());
- // TODO implement clean chaining.
fut0.whenComplete(new BiConsumer<PeerId, Throwable>() {
@Override public void accept(PeerId peerId, Throwable error) {
if (error == null) {
- rpcClient.invokeAsync(state.getLeader().getNode(), request, new InvokeCallback<R>() {
+ rpcClient.invokeAsync(peerId.getNode(), request, new InvokeCallback<R>() {
@Override public void complete(R response, Throwable err) {
if (err != null)
fut.completeExceptionally(err);
@@ -175,4 +190,26 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return fut;
}
+
+ private static class StateImpl implements State {
+ private volatile PeerId leader;
+
+ private volatile List<PeerId> peers;
+
+ private volatile List<PeerId> learners;
+
+ private AtomicReference<CompletableFuture<GetLeaderResponse>> updateFutRef = new AtomicReference<>();
+
+ @Override public @Nullable PeerId leader() {
+ return leader;
+ }
+
+ @Override public @Nullable List<PeerId> peers() {
+ return peers;
+ }
+
+ @Override public @Nullable List<PeerId> learners() {
+ return learners;
+ }
+ }
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestListener.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestListener.java
index 3649a41..c007086 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestListener.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestListener.java
@@ -3,7 +3,7 @@ package org.apache.ignite.raft.client.service;
import java.util.Iterator;
/**
- * A listener for raft group clien requests.
+ * A listener for raft group client requests.
*/
public interface RaftGroupClientRequestListener {
void onReads(Iterator iterator);
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
index 9fdc19e..a771bc9 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
@@ -6,5 +6,12 @@ import java.util.concurrent.CompletableFuture;
*
*/
public interface RaftGroupClientRequestService {
+ /**
+ * Submits a custom request to a raft group leader. If a leader is not known yet, will try to resolve the leader.
+ * @param request
+ * @param <T> Request.
+ * @param <R> Response.
+ * @return A future.
+ */
<T, R> CompletableFuture<R> submit(T request);
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupManagementServiceImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupManagementServiceImpl.java
index ff3749f..a70c1fb 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupManagementServiceImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupManagementServiceImpl.java
@@ -3,10 +3,8 @@ package org.apache.ignite.raft.client.service.impl;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.State;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.client.service.RaftGroupManagmentService;
-import org.apache.ignite.raft.rpc.Node;
import org.jetbrains.annotations.Nullable;
public class RaftGroupManagementServiceImpl implements RaftGroupManagmentService {
@@ -17,15 +15,15 @@ public class RaftGroupManagementServiceImpl implements RaftGroupManagmentService
}
@Override public @Nullable PeerId getLeader(String groupId) {
- return rpcClient.state(groupId).getLeader();
+ return rpcClient.state(groupId).leader();
}
@Override public @Nullable List<PeerId> getPeers(String groupId) {
- return rpcClient.state(groupId).getPeers();
+ return rpcClient.state(groupId).peers();
}
@Override public @Nullable List<PeerId> getLearners(String groupId) {
- return rpcClient.state(groupId).getLearners();
+ return rpcClient.state(groupId).learners();
}
@Override public CompletableFuture<PeersChangeState> addPeer(PeerId peerId) {
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/NodeImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/NodeImpl.java
index f71424f..ea6f10c 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/NodeImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/NodeImpl.java
@@ -29,7 +29,7 @@ public class NodeImpl implements Node {
}
@Override public String id() {
- return null;
+ return id;
}
@Override public boolean equals(Object o) {
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
index 0ee4561..97502ce 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
@@ -1,5 +1,8 @@
package org.apache.ignite.raft.rpc;
+/**
+ * A message for a specific raft group.
+ */
public interface RaftGroupMessage extends Message {
public String getGroupId();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RpcClient.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RpcClient.java
index cdb6db7..4231dae 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RpcClient.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RpcClient.java
@@ -28,7 +28,7 @@ public interface RpcClient {
* @param node target address
* @return true if there is a connection and the connection is active and writable.
*/
- boolean checkConnection(final NodeImpl node);
+ boolean checkConnection(final Node node);
/**
* Check connection for given address and async to create a new one if there is no connection.
@@ -37,14 +37,14 @@ public interface RpcClient {
* @param createIfAbsent create a new one if there is no connection
* @return true if there is a connection and the connection is active and writable.
*/
- boolean checkConnection(final NodeImpl node, final boolean createIfAbsent);
+ boolean checkConnection(final Node node, final boolean createIfAbsent);
/**
* Close all connections of a address.
*
* @param node target address
*/
- void closeConnection(final NodeImpl node);
+ void closeConnection(final Node node);
/**
* Asynchronous invocation with a callback.
@@ -53,7 +53,7 @@ public interface RpcClient {
* @param request Request object
* @param callback Invoke callback.
* @param executor Executor to run invoke callback.
- * @param timeoutMs Timeout millisecond
+ * @param timeoutMs Timeout millisecond.
*/
void invokeAsync(final Node node, final Message request, InvokeCallback callback, Executor executor, final long timeoutMs);
}
diff --git a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
index e7012e2..d755e86 100644
--- a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
+++ b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
@@ -1,10 +1,14 @@
package org.apache.ignite.raft.client;
import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
-import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReferenceArray;
import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.State;
+import org.apache.ignite.raft.client.RaftClientCommonMessages.GetLeaderRequest;
import org.apache.ignite.raft.client.rpc.impl.RaftGroupRpcClientImpl;
import org.apache.ignite.raft.client.message.RaftClientCommonMessageBuilderFactory;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
@@ -15,6 +19,7 @@ import org.apache.ignite.raft.rpc.RaftGroupMessage;
import org.apache.ignite.raft.rpc.RpcClient;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.parallel.Execution;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -23,6 +28,7 @@ import org.mockito.stubbing.Answer;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
@@ -35,25 +41,104 @@ public class RaftGroupRpcClientTest {
private static PeerId leader = new PeerId(new NodeImpl("test"));
@Test
+ public void testRefreshLeader() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(false);
+
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ 5_000, Collections.singleton(leader.getNode()));
+
+ PeerId leaderId = client.refreshLeader(groupId).get();
+
+ assertEquals(leader, client.state(groupId).leader());
+ assertEquals(leader, leaderId);
+ }
+
+ @Test
+ public void testRefreshLeaderMultithreaded() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(false);
+
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ 5_000, Collections.singleton(leader.getNode()));
+
+ int cnt = 20;
+
+ Thread[] runners = new Thread[cnt];
+
+ CountDownLatch l = new CountDownLatch(cnt);
+
+ for (int i = 0; i < runners.length; i++) {
+ runners[i] = new Thread(new Runnable() {
+ @Override public void run() {
+ l.countDown();
+ try {
+ l.await();
+ }
+ catch (InterruptedException e) {
+ // Ignored.
+ }
+
+ try {
+ PeerId leaderId = client.refreshLeader(groupId).get();
+
+ assertEquals(leader, client.state(groupId).leader());
+ assertEquals(leader, leaderId);
+ }
+ catch (Exception e) {
+ fail(e);
+ }
+ }
+ });
+ runners[i].setName("Executor-" + i);
+ runners[i].start();
+ }
+
+ for (int i = 0; i < runners.length; i++)
+ runners[i].join();
+ }
+
+ @Test
+ public void testRefreshLeaderTimeout() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(true);
+
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ 5_000, Collections.singleton(leader.getNode()));
+
+ try {
+ client.refreshLeader(groupId).get();
+
+ fail();
+ }
+ catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ }
+ }
+
+ @Test
public void testCustomMessage() throws Exception {
String groupId = "test";
- mockClient();
+ mockLeaderRequest(false);
+ mockCustomRequest();
- // TODO FIXME asch where to get initial configuration for the group ?
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, 5_000, Collections.singleton(leader.getNode()));
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ 5_000, Collections.singleton(leader.getNode()));
JunkRequest req = new JunkRequest(groupId);
- Future<Message> fut = client.sendCustom(req);
+ Message resp = client.sendCustom(req).get();
State state = client.state(groupId);
// Expecting raft group state to be transparently loaded on first request.
- // TODO FIXME broken
- assertEquals(leader, state.getLeader());
+ assertEquals(leader, state.leader());
- assertTrue(fut.get() instanceof JunkResponse);
+ assertTrue(resp instanceof JunkResponse);
}
private static class JunkRequest implements RaftGroupMessage {
@@ -70,8 +155,7 @@ public class RaftGroupRpcClientTest {
private static class JunkResponse implements Message {}
- private void mockClient() {
- // Mock junk request.
+ private void mockCustomRequest() {
Mockito.doAnswer(new Answer() {
@Override public Object answer(InvocationOnMock invocation) throws Throwable {
InvokeCallback callback = invocation.getArgument(2);
@@ -82,17 +166,23 @@ public class RaftGroupRpcClientTest {
return null;
}
}).when(rpcClient).invokeAsync(eq(leader.getNode()), any(JunkRequest.class), any(), any(), anyLong());
+ }
- // Mock get leader request.
+ private void mockLeaderRequest(boolean timeout) {
Mockito.doAnswer(new Answer() {
@Override public Object answer(InvocationOnMock invocation) throws Throwable {
InvokeCallback callback = invocation.getArgument(2);
Executor executor = invocation.getArgument(3);
- executor.execute(() -> callback.complete(RaftClientCommonMessageBuilderFactory.DEFAULT.createGetLeaderResponse().setLeaderId(leader).build(), null));
+ executor.execute(() -> {
+ if (timeout)
+ callback.complete(null, new TimeoutException());
+ else
+ callback.complete(RaftClientCommonMessageBuilderFactory.DEFAULT.createGetLeaderResponse().setLeaderId(leader).build(), null);
+ });
return null;
}
- }).when(rpcClient).invokeAsync(eq(leader.getNode()), any(RaftClientCommonMessages.GetLeaderRequest.class), any(), any(), anyLong());
+ }).when(rpcClient).invokeAsync(eq(leader.getNode()), any(GetLeaderRequest.class), any(), any(), anyLong());
}
}
[ignite-3] 02/04: IGNITE-14149 wip.
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch ignite-14149
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit df9226b76c769bbeadaf91dce90cf695d83d5811
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Sat Feb 27 12:57:02 2021 +0300
IGNITE-14149 wip.
---
.../raft/client/message/ClientMessageBuilderFactory.java | 4 +++-
.../apache/ignite/raft/client/rpc/RaftGroupRpcClient.java | 10 ++++++++++
.../raft/client/rpc/impl/RaftGroupRpcClientImpl.java | 11 +++++++++--
.../service/impl/RaftGroupClientRequestServiceImpl.java | 13 ++++++++++++-
.../apache/ignite/raft/client/RaftGroupRpcClientTest.java | 14 ++++++--------
5 files changed, 40 insertions(+), 12 deletions(-)
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
index 952fbfd..b0f48cf 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
@@ -4,7 +4,7 @@ import org.apache.ignite.raft.client.RaftClientCommonMessages;
/** */
public interface ClientMessageBuilderFactory {
- public static ClientMessageBuilderFactory DEFAULT = new RaftClientCommonMessageBuilderFactory();
+ public static ClientMessageBuilderFactory DEFAULT_MESSAGE_BUILDER_FACTORY = new RaftClientCommonMessageBuilderFactory();
RaftClientCommonMessages.PingRequest.Builder createPingRequest();
@@ -43,4 +43,6 @@ public interface ClientMessageBuilderFactory {
RaftClientCommonMessages.ResetLearnersRequest.Builder createResetLearnersRequest();
RaftClientCommonMessages.LearnersOpResponse.Builder createLearnersOpResponse();
+
+ RaftClientCommonMessages.UserRequest.Builder createUserRequest();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
index 5f657eb..2eca6e1 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
@@ -19,6 +19,7 @@ package org.apache.ignite.raft.client.rpc;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.raft.State;
import org.apache.ignite.raft.PeerId;
+import org.apache.ignite.raft.client.message.ClientMessageBuilderFactory;
import org.apache.ignite.raft.rpc.Message;
import org.apache.ignite.raft.rpc.RaftGroupMessage;
import org.jetbrains.annotations.Nullable;
@@ -58,6 +59,13 @@ public interface RaftGroupRpcClient {
CompletableFuture<PeerId> refreshLeader(String groupId);
/**
+ * Refreshes group members (but without a leader).
+ * @param groupId Group id.
+ * @return A future.
+ */
+ CompletableFuture<State> refreshMembers(String groupId);
+
+ /**
* Adds a voring peer to the raft group.
*
* @param request request data
@@ -152,4 +160,6 @@ public interface RaftGroupRpcClient {
* @return a future with result
*/
<R extends Message> CompletableFuture<R> sendCustom(RaftGroupMessage request);
+
+ ClientMessageBuilderFactory factory();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
index f0ceda6..e9c67cb 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
@@ -27,7 +27,6 @@ import org.apache.ignite.raft.rpc.RpcClient;
import org.jetbrains.annotations.Nullable;
import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.apache.ignite.raft.client.message.RaftClientCommonMessageBuilderFactory.DEFAULT;
public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
private final ExecutorService executor;
@@ -84,6 +83,10 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
});
}
+ @Override public CompletableFuture<State> refreshMembers(String groupId) {
+ return null;
+ }
+
@Override public CompletableFuture<RaftClientCommonMessages.AddPeerResponse> addPeer(RaftClientCommonMessages.AddPeerRequest request) {
return null;
}
@@ -130,7 +133,7 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return fut;
if (state.updateFutRef.compareAndSet(null, (fut = new CompletableFuture<>()))) {
- RaftClientCommonMessages.GetLeaderRequest req = DEFAULT.createGetLeaderRequest().setGroupId(groupId).build();
+ RaftClientCommonMessages.GetLeaderRequest req = factory.createGetLeaderRequest().setGroupId(groupId).build();
CompletableFuture<GetLeaderResponse> finalFut = fut;
@@ -191,6 +194,10 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return fut;
}
+ @Override public ClientMessageBuilderFactory factory() {
+ return this.factory;
+ }
+
private static class StateImpl implements State {
private volatile PeerId leader;
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
index 303a90e..b5439c2 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
@@ -1,8 +1,11 @@
package org.apache.ignite.raft.client.service.impl;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.raft.client.RaftClientCommonMessages;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.client.service.RaftGroupClientRequestService;
+import org.apache.ignite.raft.rpc.Message;
public class RaftGroupClientRequestServiceImpl implements RaftGroupClientRequestService {
private RaftGroupRpcClient rpcClient;
@@ -12,6 +15,14 @@ public class RaftGroupClientRequestServiceImpl implements RaftGroupClientRequest
}
@Override public <T, R> CompletableFuture<R> submit(T request) {
- return null;
+ RaftClientCommonMessages.UserRequest r = rpcClient.factory().createUserRequest().setRequest(request).build();
+
+ return rpcClient.sendCustom(r).thenApply(new Function<Message, R>() {
+ @Override public R apply(Message message) {
+ RaftClientCommonMessages.UserResponse<R> resp = (RaftClientCommonMessages.UserResponse<R>) message;
+
+ return resp.response();
+ }
+ });
}
}
diff --git a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
index d755e86..c476422 100644
--- a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
+++ b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
@@ -5,12 +5,10 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReferenceArray;
import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.State;
import org.apache.ignite.raft.client.RaftClientCommonMessages.GetLeaderRequest;
import org.apache.ignite.raft.client.rpc.impl.RaftGroupRpcClientImpl;
-import org.apache.ignite.raft.client.message.RaftClientCommonMessageBuilderFactory;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.rpc.InvokeCallback;
import org.apache.ignite.raft.rpc.Message;
@@ -19,13 +17,13 @@ import org.apache.ignite.raft.rpc.RaftGroupMessage;
import org.apache.ignite.raft.rpc.RpcClient;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
-import org.junit.jupiter.api.parallel.Execution;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.stubbing.Answer;
+import static org.apache.ignite.raft.client.message.ClientMessageBuilderFactory.DEFAULT_MESSAGE_BUILDER_FACTORY;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -46,7 +44,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
5_000, Collections.singleton(leader.getNode()));
PeerId leaderId = client.refreshLeader(groupId).get();
@@ -61,7 +59,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
5_000, Collections.singleton(leader.getNode()));
int cnt = 20;
@@ -106,7 +104,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(true);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
5_000, Collections.singleton(leader.getNode()));
try {
@@ -126,7 +124,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
mockCustomRequest();
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, new RaftClientCommonMessageBuilderFactory(),
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
5_000, Collections.singleton(leader.getNode()));
JunkRequest req = new JunkRequest(groupId);
@@ -178,7 +176,7 @@ public class RaftGroupRpcClientTest {
if (timeout)
callback.complete(null, new TimeoutException());
else
- callback.complete(RaftClientCommonMessageBuilderFactory.DEFAULT.createGetLeaderResponse().setLeaderId(leader).build(), null);
+ callback.complete(DEFAULT_MESSAGE_BUILDER_FACTORY.createGetLeaderResponse().setLeaderId(leader).build(), null);
});
return null;
[ignite-3] 03/04: IGNITE-14149 wip.
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch ignite-14149
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit ac96caef28b5f0d21fdf1cc53be5abe89de00bf9
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Sat Feb 27 13:52:39 2021 +0300
IGNITE-14149 wip.
---
...CommonMessages.java => RaftClientMessages.java} | 6 +-
.../client/message/AddLearnersRequestImpl.java | 6 +-
.../raft/client/message/AddPeerRequestImpl.java | 6 +-
.../raft/client/message/AddPeerResponseImpl.java | 6 +-
.../raft/client/message/ChangePeerRequestImpl.java | 6 +-
.../client/message/ChangePeersResponseImpl.java | 6 +-
.../message/ClientMessageBuilderFactory.java | 44 +++----
.../raft/client/message/GetLeaderRequestImpl.java | 6 +-
.../raft/client/message/GetLeaderResponseImpl.java | 6 +-
.../raft/client/message/GetPeersRequestImpl.java | 6 +-
.../raft/client/message/GetPeersResponseImpl.java | 6 +-
.../client/message/LearnersOpResponseImpl.java | 6 +-
.../raft/client/message/PingRequestImpl.java | 6 +-
.../RaftClientCommonMessageBuilderFactory.java | 84 -------------
.../message/RaftClientMessageBuilderFactory.java | 94 +++++++++++++++
.../client/message/RemoveLearnersRequestImpl.java | 6 +-
.../raft/client/message/RemovePeerRequestImpl.java | 6 +-
.../client/message/RemovePeerResponseImpl.java | 6 +-
.../client/message/ResetLearnersRequestImpl.java | 6 +-
.../raft/client/message/ResetPeerRequestImpl.java | 6 +-
.../raft/client/message/SnapshotRequestImpl.java | 7 +-
.../raft/client/message/StatusResponseImpl.java | 6 +-
.../client/message/TransferLeaderRequestImpl.java | 6 +-
.../raft/client/message/UserRequestImpl.java | 32 +++++
.../raft/client/message/UserResponseImpl.java | 21 ++++
.../ignite/raft/client/rpc/RaftGroupRpcClient.java | 34 +++---
.../client/rpc/impl/RaftGroupRpcClientImpl.java | 26 ++--
.../client/service/RaftGroupManagmentService.java | 8 +-
.../impl/RaftGroupClientRequestServiceImpl.java | 13 +-
.../client/{ => rpc}/RaftGroupRpcClientTest.java | 25 ++--
.../service/RaftGroupClientRequestServiceTest.java | 132 +++++++++++++++++++++
31 files changed, 416 insertions(+), 218 deletions(-)
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientMessages.java
similarity index 98%
rename from modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java
rename to modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientMessages.java
index 5a40dc6..3f44685 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientCommonMessages.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftClientMessages.java
@@ -27,8 +27,8 @@ import org.apache.ignite.raft.rpc.RaftGroupMessage;
/**
*
*/
-public final class RaftClientCommonMessages {
- private RaftClientCommonMessages() {
+public final class RaftClientMessages {
+ private RaftClientMessages() {
}
public interface StatusResponse extends Message {
@@ -263,6 +263,8 @@ public final class RaftClientCommonMessages {
public interface Builder<T> {
Builder setRequest(T request);
+ Builder setGroupId(String groupId);
+
UserRequest<T> build();
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
index cc3a763..7261c54 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-public class AddLearnersRequestImpl implements RaftClientCommonMessages.AddLearnersRequest, RaftClientCommonMessages.AddLearnersRequest.Builder {
+public class AddLearnersRequestImpl implements RaftClientMessages.AddLearnersRequest, RaftClientMessages.AddLearnersRequest.Builder {
private String groupId;
private PeerId leaderId;
private List<PeerId> learnersList = new ArrayList<>();
@@ -31,7 +31,7 @@ public class AddLearnersRequestImpl implements RaftClientCommonMessages.AddLearn
return this;
}
- @Override public RaftClientCommonMessages.AddLearnersRequest build() {
+ @Override public RaftClientMessages.AddLearnersRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerRequestImpl.java
index 01fb298..ba6d2c8 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerRequestImpl.java
@@ -1,9 +1,9 @@
package org.apache.ignite.raft.client.message;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class AddPeerRequestImpl implements RaftClientCommonMessages.AddPeerRequest, RaftClientCommonMessages.AddPeerRequest.Builder {
+class AddPeerRequestImpl implements RaftClientMessages.AddPeerRequest, RaftClientMessages.AddPeerRequest.Builder {
private String groupId;
private PeerId leaderId;
private PeerId peerId;
@@ -28,7 +28,7 @@ class AddPeerRequestImpl implements RaftClientCommonMessages.AddPeerRequest, Raf
return this;
}
- @Override public RaftClientCommonMessages.AddPeerRequest build() {
+ @Override public RaftClientMessages.AddPeerRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerResponseImpl.java
index 127124a..b9e1c00 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeerResponseImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class AddPeerResponseImpl implements RaftClientCommonMessages.AddPeerResponse, RaftClientCommonMessages.AddPeerResponse.Builder {
+class AddPeerResponseImpl implements RaftClientMessages.AddPeerResponse, RaftClientMessages.AddPeerResponse.Builder {
private List<PeerId> oldPeersList = new ArrayList<>();
private List<PeerId> newPeersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class AddPeerResponseImpl implements RaftClientCommonMessages.AddPeerResponse, R
return this;
}
- @Override public RaftClientCommonMessages.AddPeerResponse build() {
+ @Override public RaftClientMessages.AddPeerResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeerRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeerRequestImpl.java
index 06570ea..47d3abc 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeerRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeerRequestImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class ChangePeerRequestImpl implements RaftClientCommonMessages.ChangePeersRequest, RaftClientCommonMessages.ChangePeersRequest.Builder {
+class ChangePeerRequestImpl implements RaftClientMessages.ChangePeersRequest, RaftClientMessages.ChangePeersRequest.Builder {
private String groupId;
private List<PeerId> newPeersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class ChangePeerRequestImpl implements RaftClientCommonMessages.ChangePeersReque
return this;
}
- @Override public RaftClientCommonMessages.ChangePeersRequest build() {
+ @Override public RaftClientMessages.ChangePeersRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponseImpl.java
index d22d8f0..bc9f74d 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponseImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class ChangePeersResponseImpl implements RaftClientCommonMessages.ChangePeersResponse, RaftClientCommonMessages.ChangePeersResponse.Builder {
+class ChangePeersResponseImpl implements RaftClientMessages.ChangePeersResponse, RaftClientMessages.ChangePeersResponse.Builder {
private List<PeerId> oldPeersList = new ArrayList<>();
private List<PeerId> newPeersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class ChangePeersResponseImpl implements RaftClientCommonMessages.ChangePeersRes
return this;
}
- @Override public RaftClientCommonMessages.ChangePeersResponse build() {
+ @Override public RaftClientMessages.ChangePeersResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
index b0f48cf..4351a67 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ClientMessageBuilderFactory.java
@@ -1,48 +1,48 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
/** */
public interface ClientMessageBuilderFactory {
- public static ClientMessageBuilderFactory DEFAULT_MESSAGE_BUILDER_FACTORY = new RaftClientCommonMessageBuilderFactory();
+ RaftClientMessages.PingRequest.Builder createPingRequest();
- RaftClientCommonMessages.PingRequest.Builder createPingRequest();
+ RaftClientMessages.StatusResponse.Builder createStatusResponse();
- RaftClientCommonMessages.StatusResponse.Builder createStatusResponse();
+ RaftClientMessages.AddPeerRequest.Builder createAddPeerRequest();
- RaftClientCommonMessages.AddPeerRequest.Builder createAddPeerRequest();
+ RaftClientMessages.AddPeerResponse.Builder createAddPeerResponse();
- RaftClientCommonMessages.AddPeerResponse.Builder createAddPeerResponse();
+ RaftClientMessages.RemovePeerRequest.Builder createRemovePeerRequest();
- RaftClientCommonMessages.RemovePeerRequest.Builder createRemovePeerRequest();
+ RaftClientMessages.RemovePeerResponse.Builder createRemovePeerResponse();
- RaftClientCommonMessages.RemovePeerResponse.Builder createRemovePeerResponse();
+ RaftClientMessages.ChangePeersRequest.Builder createChangePeerRequest();
- RaftClientCommonMessages.ChangePeersRequest.Builder createChangePeerRequest();
+ RaftClientMessages.ChangePeersResponse.Builder createChangePeerResponse();
- RaftClientCommonMessages.ChangePeersResponse.Builder createChangePeerResponse();
+ RaftClientMessages.SnapshotRequest.Builder createSnapshotRequest();
- RaftClientCommonMessages.SnapshotRequest.Builder createSnapshotRequest();
+ RaftClientMessages.ResetPeerRequest.Builder createResetPeerRequest();
- RaftClientCommonMessages.ResetPeerRequest.Builder createResetPeerRequest();
+ RaftClientMessages.TransferLeaderRequest.Builder createTransferLeaderRequest();
- RaftClientCommonMessages.TransferLeaderRequest.Builder createTransferLeaderRequest();
+ RaftClientMessages.GetLeaderRequest.Builder createGetLeaderRequest();
- RaftClientCommonMessages.GetLeaderRequest.Builder createGetLeaderRequest();
+ RaftClientMessages.GetLeaderResponse.Builder createGetLeaderResponse();
- RaftClientCommonMessages.GetLeaderResponse.Builder createGetLeaderResponse();
+ RaftClientMessages.GetPeersRequest.Builder createGetPeersRequest();
- RaftClientCommonMessages.GetPeersRequest.Builder createGetPeersRequest();
+ RaftClientMessages.GetPeersResponse.Builder createGetPeersResponse();
- RaftClientCommonMessages.GetPeersResponse.Builder createGetPeersResponse();
+ RaftClientMessages.AddLearnersRequest.Builder createAddLearnersRequest();
- RaftClientCommonMessages.AddLearnersRequest.Builder createAddLearnersRequest();
+ RaftClientMessages.RemoveLearnersRequest.Builder createRemoveLearnersRequest();
- RaftClientCommonMessages.RemoveLearnersRequest.Builder createRemoveLearnersRequest();
+ RaftClientMessages.ResetLearnersRequest.Builder createResetLearnersRequest();
- RaftClientCommonMessages.ResetLearnersRequest.Builder createResetLearnersRequest();
+ RaftClientMessages.LearnersOpResponse.Builder createLearnersOpResponse();
- RaftClientCommonMessages.LearnersOpResponse.Builder createLearnersOpResponse();
+ RaftClientMessages.UserRequest.Builder createUserRequest();
- RaftClientCommonMessages.UserRequest.Builder createUserRequest();
+ RaftClientMessages.UserResponse.Builder createUserResponse();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequestImpl.java
index 429bab4..dbc7566 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequestImpl.java
@@ -1,9 +1,9 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-public class GetLeaderRequestImpl implements RaftClientCommonMessages.GetLeaderRequest, RaftClientCommonMessages.GetLeaderRequest.Builder {
+public class GetLeaderRequestImpl implements RaftClientMessages.GetLeaderRequest, RaftClientMessages.GetLeaderRequest.Builder {
private String groupId;
@Override public String getGroupId() {
@@ -16,7 +16,7 @@ public class GetLeaderRequestImpl implements RaftClientCommonMessages.GetLeaderR
return this;
}
- @Override public RaftClientCommonMessages.GetLeaderRequest build() {
+ @Override public RaftClientMessages.GetLeaderRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponseImpl.java
index 22789b1..0c84245 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponseImpl.java
@@ -2,16 +2,16 @@ package org.apache.ignite.raft.client.message;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-public class GetLeaderResponseImpl implements RaftClientCommonMessages.GetLeaderResponse, RaftClientCommonMessages.GetLeaderResponse.Builder {
+public class GetLeaderResponseImpl implements RaftClientMessages.GetLeaderResponse, RaftClientMessages.GetLeaderResponse.Builder {
private PeerId leaderId;
@Override public PeerId getLeaderId() {
return leaderId;
}
- @Override public RaftClientCommonMessages.GetLeaderResponse build() {
+ @Override public RaftClientMessages.GetLeaderResponse build() {
return this;
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequestImpl.java
index 1851152..ea50a57 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequestImpl.java
@@ -1,8 +1,8 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class GetPeersRequestImpl implements RaftClientCommonMessages.GetPeersRequest, RaftClientCommonMessages.GetPeersRequest.Builder {
+class GetPeersRequestImpl implements RaftClientMessages.GetPeersRequest, RaftClientMessages.GetPeersRequest.Builder {
private String groupId;
private boolean onlyAlive;
@@ -26,7 +26,7 @@ class GetPeersRequestImpl implements RaftClientCommonMessages.GetPeersRequest, R
return this;
}
- @Override public RaftClientCommonMessages.GetPeersRequest build() {
+ @Override public RaftClientMessages.GetPeersRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponseImpl.java
index 37a4912..bed9b6e 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponseImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class GetPeersResponseImpl implements RaftClientCommonMessages.GetPeersResponse, RaftClientCommonMessages.GetPeersResponse.Builder {
+class GetPeersResponseImpl implements RaftClientMessages.GetPeersResponse, RaftClientMessages.GetPeersResponse.Builder {
private List<PeerId> peersList = new ArrayList<>();
private List<PeerId> learnersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class GetPeersResponseImpl implements RaftClientCommonMessages.GetPeersResponse,
return this;
}
- @Override public RaftClientCommonMessages.GetPeersResponse build() {
+ @Override public RaftClientMessages.GetPeersResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/LearnersOpResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/LearnersOpResponseImpl.java
index 4c3f4c1..b929e90 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/LearnersOpResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/LearnersOpResponseImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class LearnersOpResponseImpl implements RaftClientCommonMessages.LearnersOpResponse, RaftClientCommonMessages.LearnersOpResponse.Builder {
+class LearnersOpResponseImpl implements RaftClientMessages.LearnersOpResponse, RaftClientMessages.LearnersOpResponse.Builder {
private List<PeerId> oldLearnersList = new ArrayList<>();
private List<PeerId> newLearnersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class LearnersOpResponseImpl implements RaftClientCommonMessages.LearnersOpRespo
return this;
}
- @Override public RaftClientCommonMessages.LearnersOpResponse build() {
+ @Override public RaftClientMessages.LearnersOpResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/PingRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/PingRequestImpl.java
index d4ad03d..b0c0a78 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/PingRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/PingRequestImpl.java
@@ -1,8 +1,8 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class PingRequestImpl implements RaftClientCommonMessages.PingRequest, RaftClientCommonMessages.PingRequest.Builder {
+class PingRequestImpl implements RaftClientMessages.PingRequest, RaftClientMessages.PingRequest.Builder {
private long sendTimestamp;
@Override public long getSendTimestamp() {
@@ -15,7 +15,7 @@ class PingRequestImpl implements RaftClientCommonMessages.PingRequest, RaftClien
return this;
}
- @Override public RaftClientCommonMessages.PingRequest build() {
+ @Override public RaftClientMessages.PingRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
deleted file mode 100644
index 440cee2..0000000
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientCommonMessageBuilderFactory.java
+++ /dev/null
@@ -1,84 +0,0 @@
-package org.apache.ignite.raft.client.message;
-
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
-
-/**
- * Raft client message builders factory.
- */
-public class RaftClientCommonMessageBuilderFactory implements ClientMessageBuilderFactory {
- @Override public RaftClientCommonMessages.PingRequest.Builder createPingRequest() {
- return new PingRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.StatusResponse.Builder createStatusResponse() {
- return new StatusResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.AddPeerRequest.Builder createAddPeerRequest() {
- return new AddPeerRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.AddPeerResponse.Builder createAddPeerResponse() {
- return new AddPeerResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.RemovePeerRequest.Builder createRemovePeerRequest() {
- return new RemovePeerRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.RemovePeerResponse.Builder createRemovePeerResponse() {
- return new RemovePeerResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.ChangePeersRequest.Builder createChangePeerRequest() {
- return new ChangePeerRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.ChangePeersResponse.Builder createChangePeerResponse() {
- return new ChangePeersResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.SnapshotRequest.Builder createSnapshotRequest() {
- return new SnapshotRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.ResetPeerRequest.Builder createResetPeerRequest() {
- return new ResetPeerRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.TransferLeaderRequest.Builder createTransferLeaderRequest() {
- return new TransferLeaderRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.GetLeaderRequest.Builder createGetLeaderRequest() {
- return new GetLeaderRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.GetLeaderResponse.Builder createGetLeaderResponse() {
- return new GetLeaderResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.GetPeersRequest.Builder createGetPeersRequest() {
- return new GetPeersRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.GetPeersResponse.Builder createGetPeersResponse() {
- return new GetPeersResponseImpl();
- }
-
- @Override public RaftClientCommonMessages.AddLearnersRequest.Builder createAddLearnersRequest() {
- return new AddLearnersRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.RemoveLearnersRequest.Builder createRemoveLearnersRequest() {
- return new RemoveLearnersRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.ResetLearnersRequest.Builder createResetLearnersRequest() {
- return new ResetLearnersRequestImpl();
- }
-
- @Override public RaftClientCommonMessages.LearnersOpResponse.Builder createLearnersOpResponse() {
- return new LearnersOpResponseImpl();
- }
-}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientMessageBuilderFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientMessageBuilderFactory.java
new file mode 100644
index 0000000..2d47315
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftClientMessageBuilderFactory.java
@@ -0,0 +1,94 @@
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.RaftClientMessages;
+
+/**
+ * Raft client message builders factory.
+ */
+public class RaftClientMessageBuilderFactory implements ClientMessageBuilderFactory {
+ public static RaftClientMessageBuilderFactory INSTANCE = new RaftClientMessageBuilderFactory();
+
+ @Override public RaftClientMessages.PingRequest.Builder createPingRequest() {
+ return new PingRequestImpl();
+ }
+
+ @Override public RaftClientMessages.StatusResponse.Builder createStatusResponse() {
+ return new StatusResponseImpl();
+ }
+
+ @Override public RaftClientMessages.AddPeerRequest.Builder createAddPeerRequest() {
+ return new AddPeerRequestImpl();
+ }
+
+ @Override public RaftClientMessages.AddPeerResponse.Builder createAddPeerResponse() {
+ return new AddPeerResponseImpl();
+ }
+
+ @Override public RaftClientMessages.RemovePeerRequest.Builder createRemovePeerRequest() {
+ return new RemovePeerRequestImpl();
+ }
+
+ @Override public RaftClientMessages.RemovePeerResponse.Builder createRemovePeerResponse() {
+ return new RemovePeerResponseImpl();
+ }
+
+ @Override public RaftClientMessages.ChangePeersRequest.Builder createChangePeerRequest() {
+ return new ChangePeerRequestImpl();
+ }
+
+ @Override public RaftClientMessages.ChangePeersResponse.Builder createChangePeerResponse() {
+ return new ChangePeersResponseImpl();
+ }
+
+ @Override public RaftClientMessages.SnapshotRequest.Builder createSnapshotRequest() {
+ return new SnapshotRequestImpl();
+ }
+
+ @Override public RaftClientMessages.ResetPeerRequest.Builder createResetPeerRequest() {
+ return new ResetPeerRequestImpl();
+ }
+
+ @Override public RaftClientMessages.TransferLeaderRequest.Builder createTransferLeaderRequest() {
+ return new TransferLeaderRequestImpl();
+ }
+
+ @Override public RaftClientMessages.GetLeaderRequest.Builder createGetLeaderRequest() {
+ return new GetLeaderRequestImpl();
+ }
+
+ @Override public RaftClientMessages.GetLeaderResponse.Builder createGetLeaderResponse() {
+ return new GetLeaderResponseImpl();
+ }
+
+ @Override public RaftClientMessages.GetPeersRequest.Builder createGetPeersRequest() {
+ return new GetPeersRequestImpl();
+ }
+
+ @Override public RaftClientMessages.GetPeersResponse.Builder createGetPeersResponse() {
+ return new GetPeersResponseImpl();
+ }
+
+ @Override public RaftClientMessages.AddLearnersRequest.Builder createAddLearnersRequest() {
+ return new AddLearnersRequestImpl();
+ }
+
+ @Override public RaftClientMessages.RemoveLearnersRequest.Builder createRemoveLearnersRequest() {
+ return new RemoveLearnersRequestImpl();
+ }
+
+ @Override public RaftClientMessages.ResetLearnersRequest.Builder createResetLearnersRequest() {
+ return new ResetLearnersRequestImpl();
+ }
+
+ @Override public RaftClientMessages.LearnersOpResponse.Builder createLearnersOpResponse() {
+ return new LearnersOpResponseImpl();
+ }
+
+ @Override public RaftClientMessages.UserRequest.Builder createUserRequest() {
+ return new UserRequestImpl();
+ }
+
+ @Override public RaftClientMessages.UserResponse.Builder createUserResponse() {
+ return new UserResponseImpl();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequestImpl.java
index de0b056..e55aefc 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequestImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class RemoveLearnersRequestImpl implements RaftClientCommonMessages.RemoveLearnersRequest, RaftClientCommonMessages.RemoveLearnersRequest.Builder {
+class RemoveLearnersRequestImpl implements RaftClientMessages.RemoveLearnersRequest, RaftClientMessages.RemoveLearnersRequest.Builder {
private String groupId;
private List<PeerId> learnersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class RemoveLearnersRequestImpl implements RaftClientCommonMessages.RemoveLearne
return this;
}
- @Override public RaftClientCommonMessages.RemoveLearnersRequest build() {
+ @Override public RaftClientMessages.RemoveLearnersRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerRequestImpl.java
index a068f1d..0e4934d 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerRequestImpl.java
@@ -1,9 +1,9 @@
package org.apache.ignite.raft.client.message;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class RemovePeerRequestImpl implements RaftClientCommonMessages.RemovePeerRequest, RaftClientCommonMessages.RemovePeerRequest.Builder {
+class RemovePeerRequestImpl implements RaftClientMessages.RemovePeerRequest, RaftClientMessages.RemovePeerRequest.Builder {
private String groupId;
private PeerId peerId;
@@ -27,7 +27,7 @@ class RemovePeerRequestImpl implements RaftClientCommonMessages.RemovePeerReques
return this;
}
- @Override public RaftClientCommonMessages.RemovePeerRequest build() {
+ @Override public RaftClientMessages.RemovePeerRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerResponseImpl.java
index 35d9013..00eae18 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeerResponseImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class RemovePeerResponseImpl implements RaftClientCommonMessages.RemovePeerResponse, RaftClientCommonMessages.RemovePeerResponse.Builder {
+class RemovePeerResponseImpl implements RaftClientMessages.RemovePeerResponse, RaftClientMessages.RemovePeerResponse.Builder {
private List<PeerId> oldPeersList = new ArrayList<>();
private List<PeerId> newPeersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class RemovePeerResponseImpl implements RaftClientCommonMessages.RemovePeerRespo
return this;
}
- @Override public RaftClientCommonMessages.RemovePeerResponse build() {
+ @Override public RaftClientMessages.RemovePeerResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetLearnersRequestImpl.java
index 1e5188d..340a39f 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetLearnersRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetLearnersRequestImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class ResetLearnersRequestImpl implements RaftClientCommonMessages.ResetLearnersRequest, RaftClientCommonMessages.ResetLearnersRequest.Builder {
+class ResetLearnersRequestImpl implements RaftClientMessages.ResetLearnersRequest, RaftClientMessages.ResetLearnersRequest.Builder {
private String groupId;
private List<PeerId> learnersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class ResetLearnersRequestImpl implements RaftClientCommonMessages.ResetLearners
return this;
}
- @Override public RaftClientCommonMessages.ResetLearnersRequest build() {
+ @Override public RaftClientMessages.ResetLearnersRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetPeerRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetPeerRequestImpl.java
index bb9c28b..6c8d4bc 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetPeerRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ResetPeerRequestImpl.java
@@ -3,9 +3,9 @@ package org.apache.ignite.raft.client.message;
import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class ResetPeerRequestImpl implements RaftClientCommonMessages.ResetPeerRequest, RaftClientCommonMessages.ResetPeerRequest.Builder {
+class ResetPeerRequestImpl implements RaftClientMessages.ResetPeerRequest, RaftClientMessages.ResetPeerRequest.Builder {
private String groupId;
private List<PeerId> newPeersList = new ArrayList<>();
@@ -29,7 +29,7 @@ class ResetPeerRequestImpl implements RaftClientCommonMessages.ResetPeerRequest,
return this;
}
- @Override public RaftClientCommonMessages.ResetPeerRequest build() {
+ @Override public RaftClientMessages.ResetPeerRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequestImpl.java
index 5f1bde1..08a09dc 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequestImpl.java
@@ -1,9 +1,8 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class SnapshotRequestImpl implements RaftClientCommonMessages.SnapshotRequest, RaftClientCommonMessages.SnapshotRequest.Builder {
+class SnapshotRequestImpl implements RaftClientMessages.SnapshotRequest, RaftClientMessages.SnapshotRequest.Builder {
private String groupId;
@Override public String getGroupId() {
@@ -16,7 +15,7 @@ class SnapshotRequestImpl implements RaftClientCommonMessages.SnapshotRequest, R
return this;
}
- @Override public RaftClientCommonMessages.SnapshotRequest build() {
+ @Override public RaftClientMessages.SnapshotRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/StatusResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/StatusResponseImpl.java
index 5afcf22..6ae0b6b 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/StatusResponseImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/StatusResponseImpl.java
@@ -1,8 +1,8 @@
package org.apache.ignite.raft.client.message;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class StatusResponseImpl implements RaftClientCommonMessages.StatusResponse, RaftClientCommonMessages.StatusResponse.Builder {
+class StatusResponseImpl implements RaftClientMessages.StatusResponse, RaftClientMessages.StatusResponse.Builder {
private int errorCode;
private String errorMsg = "";
@@ -26,7 +26,7 @@ class StatusResponseImpl implements RaftClientCommonMessages.StatusResponse, Raf
return this;
}
- @Override public RaftClientCommonMessages.StatusResponse build() {
+ @Override public RaftClientMessages.StatusResponse build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeaderRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeaderRequestImpl.java
index b125aec..2cc6592 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeaderRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeaderRequestImpl.java
@@ -1,9 +1,9 @@
package org.apache.ignite.raft.client.message;
import org.apache.ignite.raft.PeerId;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
-class TransferLeaderRequestImpl implements RaftClientCommonMessages.TransferLeaderRequest, RaftClientCommonMessages.TransferLeaderRequest.Builder {
+class TransferLeaderRequestImpl implements RaftClientMessages.TransferLeaderRequest, RaftClientMessages.TransferLeaderRequest.Builder {
private String groupId;
private PeerId peerId;
@@ -21,7 +21,7 @@ class TransferLeaderRequestImpl implements RaftClientCommonMessages.TransferLead
return this;
}
- @Override public RaftClientCommonMessages.TransferLeaderRequest build() {
+ @Override public RaftClientMessages.TransferLeaderRequest build() {
return this;
}
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserRequestImpl.java
new file mode 100644
index 0000000..06d667e
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserRequestImpl.java
@@ -0,0 +1,32 @@
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.RaftClientMessages;
+
+public class UserRequestImpl<T> implements RaftClientMessages.UserRequest<T>, RaftClientMessages.UserRequest.Builder<T> {
+ private T request;
+ private String groupId;
+
+ @Override public T request() {
+ return request;
+ }
+
+ @Override public Builder setGroupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ @Override public Builder setRequest(T request) {
+ this.request = request;
+
+ return this;
+ }
+
+ @Override public RaftClientMessages.UserRequest<T> build() {
+ return this;
+ }
+
+ @Override public String getGroupId() {
+ return groupId;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserResponseImpl.java
new file mode 100644
index 0000000..0d62953
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/UserResponseImpl.java
@@ -0,0 +1,21 @@
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.RaftClientMessages;
+
+public class UserResponseImpl<T> implements RaftClientMessages.UserResponse<T>, RaftClientMessages.UserResponse.Builder<T> {
+ private T response;
+
+ @Override public T response() {
+ return response;
+ }
+
+ @Override public Builder setResponse(T response) {
+ this.response = response;
+
+ return this;
+ }
+
+ @Override public RaftClientMessages.UserResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
index 2eca6e1..14384e9 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClient.java
@@ -22,22 +22,21 @@ import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.client.message.ClientMessageBuilderFactory;
import org.apache.ignite.raft.rpc.Message;
import org.apache.ignite.raft.rpc.RaftGroupMessage;
-import org.jetbrains.annotations.Nullable;
-
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.AddLearnersRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.AddPeerRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.AddPeerResponse;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.ChangePeersRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.ChangePeersResponse;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.LearnersOpResponse;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.RemoveLearnersRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.RemovePeerRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.RemovePeerResponse;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.ResetLearnersRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.ResetPeerRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.SnapshotRequest;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.StatusResponse;
-import static org.apache.ignite.raft.client.RaftClientCommonMessages.TransferLeaderRequest;
+
+import static org.apache.ignite.raft.client.RaftClientMessages.AddLearnersRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.AddPeerRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.AddPeerResponse;
+import static org.apache.ignite.raft.client.RaftClientMessages.ChangePeersRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.ChangePeersResponse;
+import static org.apache.ignite.raft.client.RaftClientMessages.LearnersOpResponse;
+import static org.apache.ignite.raft.client.RaftClientMessages.RemoveLearnersRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.RemovePeerRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.RemovePeerResponse;
+import static org.apache.ignite.raft.client.RaftClientMessages.ResetLearnersRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.ResetPeerRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.SnapshotRequest;
+import static org.apache.ignite.raft.client.RaftClientMessages.StatusResponse;
+import static org.apache.ignite.raft.client.RaftClientMessages.TransferLeaderRequest;
/**
* Low-level raft group RPC client.
@@ -161,5 +160,8 @@ public interface RaftGroupRpcClient {
*/
<R extends Message> CompletableFuture<R> sendCustom(RaftGroupMessage request);
+ /**
+ * @return A message builder factory.
+ */
ClientMessageBuilderFactory factory();
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
index e9c67cb..4599556 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/rpc/impl/RaftGroupRpcClientImpl.java
@@ -14,9 +14,9 @@ import java.util.function.BiConsumer;
import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.RaftException;
import org.apache.ignite.raft.State;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
-import org.apache.ignite.raft.client.RaftClientCommonMessages.GetLeaderResponse;
-import org.apache.ignite.raft.client.RaftClientCommonMessages.StatusResponse;
+import org.apache.ignite.raft.client.RaftClientMessages;
+import org.apache.ignite.raft.client.RaftClientMessages.GetLeaderResponse;
+import org.apache.ignite.raft.client.RaftClientMessages.StatusResponse;
import org.apache.ignite.raft.client.message.ClientMessageBuilderFactory;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.rpc.InvokeCallback;
@@ -87,39 +87,39 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.AddPeerResponse> addPeer(RaftClientCommonMessages.AddPeerRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.AddPeerResponse> addPeer(RaftClientMessages.AddPeerRequest request) {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.RemovePeerResponse> removePeer(RaftClientCommonMessages.RemovePeerRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.RemovePeerResponse> removePeer(RaftClientMessages.RemovePeerRequest request) {
return null;
}
- @Override public CompletableFuture<StatusResponse> resetPeers(PeerId peerId, RaftClientCommonMessages.ResetPeerRequest request) {
+ @Override public CompletableFuture<StatusResponse> resetPeers(PeerId peerId, RaftClientMessages.ResetPeerRequest request) {
return null;
}
- @Override public CompletableFuture<StatusResponse> snapshot(PeerId peerId, RaftClientCommonMessages.SnapshotRequest request) {
+ @Override public CompletableFuture<StatusResponse> snapshot(PeerId peerId, RaftClientMessages.SnapshotRequest request) {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.ChangePeersResponse> changePeers(RaftClientCommonMessages.ChangePeersRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.ChangePeersResponse> changePeers(RaftClientMessages.ChangePeersRequest request) {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.LearnersOpResponse> addLearners(RaftClientCommonMessages.AddLearnersRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.LearnersOpResponse> addLearners(RaftClientMessages.AddLearnersRequest request) {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.LearnersOpResponse> removeLearners(RaftClientCommonMessages.RemoveLearnersRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.LearnersOpResponse> removeLearners(RaftClientMessages.RemoveLearnersRequest request) {
return null;
}
- @Override public CompletableFuture<RaftClientCommonMessages.LearnersOpResponse> resetLearners(RaftClientCommonMessages.ResetLearnersRequest request) {
+ @Override public CompletableFuture<RaftClientMessages.LearnersOpResponse> resetLearners(RaftClientMessages.ResetLearnersRequest request) {
return null;
}
- @Override public CompletableFuture<StatusResponse> transferLeader(RaftClientCommonMessages.TransferLeaderRequest request) {
+ @Override public CompletableFuture<StatusResponse> transferLeader(RaftClientMessages.TransferLeaderRequest request) {
return null;
}
@@ -133,7 +133,7 @@ public class RaftGroupRpcClientImpl implements RaftGroupRpcClient {
return fut;
if (state.updateFutRef.compareAndSet(null, (fut = new CompletableFuture<>()))) {
- RaftClientCommonMessages.GetLeaderRequest req = factory.createGetLeaderRequest().setGroupId(groupId).build();
+ RaftClientMessages.GetLeaderRequest req = factory.createGetLeaderRequest().setGroupId(groupId).build();
CompletableFuture<GetLeaderResponse> finalFut = fut;
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupManagmentService.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupManagmentService.java
index 725d3e1..d868dc9 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupManagmentService.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupManagmentService.java
@@ -9,24 +9,24 @@ import org.jetbrains.annotations.Nullable;
public interface RaftGroupManagmentService {
/**
* @param groupId
- * @return Peer id.
+ * @return Leader id or null if it has not been yet initialized.
*/
@Nullable PeerId getLeader(String groupId);
/**
* @param groupId
- * @return List of peers.
+ * @return List of peers or null if it has not been yet initialized.
*/
@Nullable List<PeerId> getPeers(String groupId);
/**
* @param groupId
- * @return List of peers.
+ * @return List of peers or null if it has not been yet initialized.
*/
@Nullable List<PeerId> getLearners(String groupId);
/**
- * Adds a voring peer to the raft group.
+ * Adds a voting peer to the raft group.
*
* @param request request data
* @return A future with the result
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
index b5439c2..16d63c4 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupClientRequestServiceImpl.java
@@ -2,24 +2,27 @@ package org.apache.ignite.raft.client.service.impl;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
-import org.apache.ignite.raft.client.RaftClientCommonMessages;
+import org.apache.ignite.raft.client.RaftClientMessages;
import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.client.service.RaftGroupClientRequestService;
import org.apache.ignite.raft.rpc.Message;
public class RaftGroupClientRequestServiceImpl implements RaftGroupClientRequestService {
- private RaftGroupRpcClient rpcClient;
+ private final RaftGroupRpcClient rpcClient;
+ private final String groupId;
- public RaftGroupClientRequestServiceImpl(RaftGroupRpcClient rpcClient) {
+ public RaftGroupClientRequestServiceImpl(RaftGroupRpcClient rpcClient, String groupId) {
this.rpcClient = rpcClient;
+ this.groupId = groupId;
}
@Override public <T, R> CompletableFuture<R> submit(T request) {
- RaftClientCommonMessages.UserRequest r = rpcClient.factory().createUserRequest().setRequest(request).build();
+ RaftClientMessages.UserRequest r =
+ rpcClient.factory().createUserRequest().setRequest(request).setGroupId(groupId).build();
return rpcClient.sendCustom(r).thenApply(new Function<Message, R>() {
@Override public R apply(Message message) {
- RaftClientCommonMessages.UserResponse<R> resp = (RaftClientCommonMessages.UserResponse<R>) message;
+ RaftClientMessages.UserResponse<R> resp = (RaftClientMessages.UserResponse<R>) message;
return resp.response();
}
diff --git a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClientTest.java
similarity index 86%
rename from modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
rename to modules/raft-client/src/test/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClientTest.java
index c476422..5247c59 100644
--- a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/RaftGroupRpcClientTest.java
+++ b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/rpc/RaftGroupRpcClientTest.java
@@ -1,15 +1,13 @@
-package org.apache.ignite.raft.client;
+package org.apache.ignite.raft.client.rpc;
-import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeoutException;
import org.apache.ignite.raft.PeerId;
import org.apache.ignite.raft.State;
-import org.apache.ignite.raft.client.RaftClientCommonMessages.GetLeaderRequest;
+import org.apache.ignite.raft.client.RaftClientMessages.GetLeaderRequest;
import org.apache.ignite.raft.client.rpc.impl.RaftGroupRpcClientImpl;
-import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
import org.apache.ignite.raft.rpc.InvokeCallback;
import org.apache.ignite.raft.rpc.Message;
import org.apache.ignite.raft.rpc.NodeImpl;
@@ -23,7 +21,8 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.stubbing.Answer;
-import static org.apache.ignite.raft.client.message.ClientMessageBuilderFactory.DEFAULT_MESSAGE_BUILDER_FACTORY;
+import static java.util.Collections.singleton;
+import static org.apache.ignite.raft.client.message.RaftClientMessageBuilderFactory.INSTANCE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -44,8 +43,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
- 5_000, Collections.singleton(leader.getNode()));
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, INSTANCE, 5_000, singleton(leader.getNode()));
PeerId leaderId = client.refreshLeader(groupId).get();
@@ -59,8 +57,7 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
- 5_000, Collections.singleton(leader.getNode()));
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, INSTANCE, 5_000, singleton(leader.getNode()));
int cnt = 20;
@@ -104,8 +101,8 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(true);
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
- 5_000, Collections.singleton(leader.getNode()));
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, INSTANCE,
+ 5_000, singleton(leader.getNode()));
try {
client.refreshLeader(groupId).get();
@@ -124,8 +121,8 @@ public class RaftGroupRpcClientTest {
mockLeaderRequest(false);
mockCustomRequest();
- RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, DEFAULT_MESSAGE_BUILDER_FACTORY,
- 5_000, Collections.singleton(leader.getNode()));
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, INSTANCE,
+ 5_000, singleton(leader.getNode()));
JunkRequest req = new JunkRequest(groupId);
@@ -176,7 +173,7 @@ public class RaftGroupRpcClientTest {
if (timeout)
callback.complete(null, new TimeoutException());
else
- callback.complete(DEFAULT_MESSAGE_BUILDER_FACTORY.createGetLeaderResponse().setLeaderId(leader).build(), null);
+ callback.complete(INSTANCE.createGetLeaderResponse().setLeaderId(leader).build(), null);
});
return null;
diff --git a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestServiceTest.java b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestServiceTest.java
new file mode 100644
index 0000000..56e3ff8
--- /dev/null
+++ b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestServiceTest.java
@@ -0,0 +1,132 @@
+package org.apache.ignite.raft.client.service;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import org.apache.ignite.raft.PeerId;
+import org.apache.ignite.raft.client.RaftClientMessages;
+import org.apache.ignite.raft.client.RaftClientMessages.UserRequest;
+import org.apache.ignite.raft.client.rpc.RaftGroupRpcClient;
+import org.apache.ignite.raft.client.rpc.impl.RaftGroupRpcClientImpl;
+import org.apache.ignite.raft.client.service.impl.RaftGroupClientRequestServiceImpl;
+import org.apache.ignite.raft.rpc.InvokeCallback;
+import org.apache.ignite.raft.rpc.NodeImpl;
+import org.apache.ignite.raft.rpc.RpcClient;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentMatcher;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.stubbing.Answer;
+
+import static java.util.Collections.singleton;
+import static org.apache.ignite.raft.client.message.RaftClientMessageBuilderFactory.INSTANCE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+
+@ExtendWith(MockitoExtension.class)
+public class RaftGroupClientRequestServiceTest {
+ @Mock
+ private RpcClient rpcClient;
+
+ private static PeerId leader = new PeerId(new NodeImpl("test"));
+
+ @Test
+ public void testUserRequest() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest();
+ mockUserRequest1();
+ mockUserRequest2();
+
+ RaftGroupRpcClient client = new RaftGroupRpcClientImpl(rpcClient, INSTANCE, 5_000, singleton(leader.getNode()));
+
+ RaftGroupClientRequestService service = new RaftGroupClientRequestServiceImpl(client, groupId);
+
+ assertNull(client.state(groupId).leader());
+
+ CompletableFuture<TestOutput1> fut1 = service.submit(new TestInput1());
+
+ TestOutput1 output1 = fut1.get();
+
+ assertNotNull(output1);
+
+ CompletableFuture<TestOutput2> fut2 = service.submit(new TestInput2());
+
+ TestOutput2 output2 = fut2.get();
+
+ assertNotNull(output2);
+
+ assertEquals(leader, client.state(groupId).leader());
+ }
+
+ private static class TestInput1 {
+ }
+
+ private static class TestOutput1 {
+ }
+
+ private static class TestInput2 {
+ }
+
+ private static class TestOutput2 {
+ }
+
+ private void mockLeaderRequest() {
+ Mockito.doAnswer(new Answer() {
+ @Override public Object answer(InvocationOnMock invocation) throws Throwable {
+ InvokeCallback callback = invocation.getArgument(2);
+ Executor executor = invocation.getArgument(3);
+
+ executor.execute(() -> {
+ callback.complete(INSTANCE.createGetLeaderResponse().setLeaderId(leader).build(), null);
+ });
+
+ return null;
+ }
+ }).when(rpcClient).invokeAsync(eq(leader.getNode()), any(RaftClientMessages.GetLeaderRequest.class),
+ any(), any(), anyLong());
+ }
+
+ private void mockUserRequest1() {
+ Mockito.doAnswer(new Answer() {
+ @Override public Object answer(InvocationOnMock invocation) throws Throwable {
+ InvokeCallback callback = invocation.getArgument(2);
+ Executor executor = invocation.getArgument(3);
+
+ executor.execute(() -> callback.complete(INSTANCE.createUserResponse().
+ setResponse(new TestOutput1()).build(), null));
+
+ return null;
+ }
+ }).when(rpcClient).invokeAsync(eq(leader.getNode()), argThat(new ArgumentMatcher<UserRequest>() {
+ @Override public boolean matches(UserRequest argument) {
+ return argument.request() instanceof TestInput1;
+ }
+ }), any(), any(), anyLong());
+ }
+
+ private void mockUserRequest2() {
+ Mockito.doAnswer(new Answer() {
+ @Override public Object answer(InvocationOnMock invocation) throws Throwable {
+ InvokeCallback callback = invocation.getArgument(2);
+ Executor executor = invocation.getArgument(3);
+
+ executor.execute(() -> callback.complete(INSTANCE.createUserResponse().
+ setResponse(new TestOutput2()).build(), null));
+
+ return null;
+ }
+ }).when(rpcClient).invokeAsync(eq(leader.getNode()), argThat(new ArgumentMatcher<UserRequest>() {
+ @Override public boolean matches(UserRequest argument) {
+ return argument.request() instanceof TestInput2;
+ }
+ }), any(), any(), anyLong());
+ }
+}
[ignite-3] 04/04: IGNITE-14149 wip.
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch ignite-14149
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit c4871cec894703a0cc442e628e6b96953088cc55
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Sat Feb 27 14:23:31 2021 +0300
IGNITE-14149 wip.
---
.../org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java | 1 -
.../ignite/raft/client/service/RaftGroupClientRequestService.java | 2 +-
.../src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java | 2 +-
3 files changed, 2 insertions(+), 3 deletions(-)
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
index 7261c54..2bb1125 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequestImpl.java
@@ -24,7 +24,6 @@ public class AddLearnersRequestImpl implements RaftClientMessages.AddLearnersReq
return this;
}
-
@Override public Builder addLearners(PeerId learnerId) {
learnersList.add(learnerId);
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
index a771bc9..2c5e04a 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupClientRequestService.java
@@ -7,7 +7,7 @@ import java.util.concurrent.CompletableFuture;
*/
public interface RaftGroupClientRequestService {
/**
- * Submits a custom request to a raft group leader. If a leader is not known yet, will try to resolve the leader.
+ * Submits a custom request to a raft group leader. If a leader is not initialized yet, will try to resolve it.
* @param request
* @param <T> Request.
* @param <R> Response.
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
index 97502ce..071360c 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/rpc/RaftGroupMessage.java
@@ -1,7 +1,7 @@
package org.apache.ignite.raft.rpc;
/**
- * A message for a specific raft group.
+ * A message targeted to a specific raft group.
*/
public interface RaftGroupMessage extends Message {
public String getGroupId();