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/03/26 14:10:56 UTC
[ignite-3] branch main updated: IGNITE-14149 RAFT client module.
(#59)
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new cb518de IGNITE-14149 RAFT client module. (#59)
cb518de is described below
commit cb518dea2d7b6c84aa8ee9c5c9cc6720db4a75c9
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Fri Mar 26 17:10:42 2021 +0300
IGNITE-14149 RAFT client module. (#59)
---
.../java/org/apache/ignite/lang/LogWrapper.java | 80 ++++
.../org/apache/ignite/network/NetworkCluster.java | 7 +-
.../network/scalecube/ScaleCubeNetworkCluster.java | 2 +-
modules/raft-client/README.md | 3 +
modules/raft-client/pom.xml | 84 ++++
.../org/apache/ignite/raft/client/Command.java | 24 ++
.../ignite/raft/client/ElectionPriority.java | 38 ++
.../java/org/apache/ignite/raft/client/Peer.java | 96 +++++
.../apache/ignite/raft/client/RaftErrorCode.java | 61 +++
.../org/apache/ignite/raft/client/ReadCommand.java | 24 ++
.../apache/ignite/raft/client/WriteCommand.java | 24 ++
.../raft/client/exception/RaftException.java | 41 ++
.../ignite/raft/client/message/ActionRequest.java | 56 +++
.../ignite/raft/client/message/ActionResponse.java | 43 +++
.../raft/client/message/AddLearnersRequest.java | 57 +++
.../raft/client/message/AddPeersRequest.java | 57 +++
.../raft/client/message/ChangePeersResponse.java | 57 +++
.../raft/client/message/GetLeaderRequest.java | 43 +++
.../raft/client/message/GetLeaderResponse.java | 45 +++
.../raft/client/message/GetPeersRequest.java | 52 +++
.../raft/client/message/GetPeersResponse.java | 57 +++
.../raft/client/message/RaftErrorResponse.java | 58 +++
.../raft/client/message/RemoveLearnersRequest.java | 57 +++
.../raft/client/message/RemovePeersRequest.java | 57 +++
.../raft/client/message/SnapshotRequest.java | 43 +++
.../client/message/TransferLeadershipRequest.java | 56 +++
.../client/message/impl/ActionRequestImpl.java | 59 +++
.../client/message/impl/ActionResponseImpl.java | 43 +++
.../message/impl/AddLearnersRequestImpl.java | 63 +++
.../client/message/impl/AddPeersRequestImpl.java | 63 +++
.../message/impl/ChangePeersResponseImpl.java | 60 +++
.../client/message/impl/GetLeaderRequestImpl.java | 43 +++
.../client/message/impl/GetLeaderResponseImpl.java | 44 +++
.../client/message/impl/GetPeersRequestImpl.java | 58 +++
.../client/message/impl/GetPeersResponseImpl.java | 59 +++
.../message/impl/RaftClientMessageFactory.java | 108 ++++++
.../message/impl/RaftClientMessageFactoryImpl.java | 108 ++++++
.../client/message/impl/RaftErrorResponseImpl.java | 59 +++
.../message/impl/RemoveLearnersRequestImpl.java | 61 +++
.../message/impl/RemovePeersRequestImpl.java | 60 +++
.../client/message/impl/SnapshotRequestImpl.java | 43 +++
.../impl/TransferLeadershipRequestImpl.java | 61 +++
.../client/service/RaftGroupCommandListener.java | 37 ++
.../raft/client/service/RaftGroupService.java | 196 ++++++++++
.../client/service/impl/RaftGroupServiceImpl.java | 389 +++++++++++++++++++
.../raft/client/service/RaftGroupServiceTest.java | 421 +++++++++++++++++++++
pom.xml | 1 +
47 files changed, 3253 insertions(+), 5 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/LogWrapper.java b/modules/core/src/main/java/org/apache/ignite/lang/LogWrapper.java
new file mode 100644
index 0000000..bfa77c2
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/lang/LogWrapper.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.lang;
+
+import java.util.Objects;
+
+import static java.lang.System.Logger.Level.DEBUG;
+import static java.lang.System.Logger.Level.ERROR;
+import static java.lang.System.Logger.Level.INFO;
+import static java.lang.System.Logger.Level.WARNING;
+
+/**
+ * Wraps system logger for more convenient access.
+ */
+public class LogWrapper {
+ /** Logger delegate. */
+ private final System.Logger log;
+
+ /**
+ * @param cls The class for a logger.
+ */
+ public LogWrapper(Class<?> cls) {
+ this.log = System.getLogger(Objects.requireNonNull(cls).getName());
+ }
+
+ /**
+ * @param msg The message.
+ * @param params Parameters.
+ */
+ public void info(String msg, Object... params) {
+ log.log(INFO, msg, params);
+ }
+
+ /**
+ * @param msg The message.
+ * @param params Parameters.
+ */
+ public void debug(String msg, Object... params) {
+ log.log(DEBUG, msg, params);
+ }
+
+ /**
+ * @param msg The message.
+ * @param params Parameters.
+ */
+ public void warn(String msg, Object... params) {
+ log.log(WARNING, msg, params);
+ }
+
+ /**
+ * @param msg The message.
+ * @param params Parameters.
+ */
+ public void error(String msg, Object... params) {
+ log.log(ERROR, msg, params);
+ }
+
+ /**
+ * @param msg The message.
+ * @param e The exception.
+ */
+ public void error(String msg, Exception e) {
+ log.log(ERROR, msg, e);
+ }
+}
diff --git a/modules/network/src/main/java/org/apache/ignite/network/NetworkCluster.java b/modules/network/src/main/java/org/apache/ignite/network/NetworkCluster.java
index f271b53..b5cffcb 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/NetworkCluster.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/NetworkCluster.java
@@ -64,15 +64,14 @@ public interface NetworkCluster {
/**
* Sends asynchronously a message with same guarantees as for {@link #send(NetworkMember, Object)} and
- * returns a response (RPC style).
+ * returns a response.
*
* @param member Network member which should receive the message.
* @param msg A message.
* @param timeout Waiting for response timeout in milliseconds.
- * @param <R> Expected response type.
- * @return A future holding the response or error if the expected response was not received.
+ * @return A future holding the response, which can be of any type.
*/
- <R> CompletableFuture<R> sendWithResponse(NetworkMember member, Object msg, long timeout);
+ CompletableFuture<?> sendWithResponse(NetworkMember member, Object msg, long timeout);
/**
* Add provider which allows to get configured handlers for different cluster events(ex. received message).
diff --git a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeNetworkCluster.java b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeNetworkCluster.java
index b985dba..4a535ec 100644
--- a/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeNetworkCluster.java
+++ b/modules/network/src/main/java/org/apache/ignite/network/scalecube/ScaleCubeNetworkCluster.java
@@ -90,7 +90,7 @@ public class ScaleCubeNetworkCluster implements NetworkCluster {
}
/** {@inheritDoc} */
- @Override public <R> CompletableFuture<R> sendWithResponse(NetworkMember member, Object msg, long timeout) {
+ @Override public CompletableFuture<?> sendWithResponse(NetworkMember member, Object msg, long timeout) {
return cluster.requestResponse(memberResolver.resolveMember(member), fromData(msg))
.timeout(ofMillis(timeout)).toFuture().thenApply(m -> m.data());
}
diff --git a/modules/raft-client/README.md b/modules/raft-client/README.md
new file mode 100644
index 0000000..69c8568
--- /dev/null
+++ b/modules/raft-client/README.md
@@ -0,0 +1,3 @@
+# Ignite raft client module.
+This module provides a service for interoperability with RAFT replication group peers.
+
diff --git a/modules/raft-client/pom.xml b/modules/raft-client/pom.xml
new file mode 100644
index 0000000..0274f42
--- /dev/null
+++ b/modules/raft-client/pom.xml
@@ -0,0 +1,84 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+
+<!--
+ POM file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-parent</artifactId>
+ <version>1</version>
+ <relativePath>../../parent/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>ignite-raft-client</artifactId>
+ <version>3.0.0-SNAPSHOT</version>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-network</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.jetbrains</groupId>
+ <artifactId>annotations</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <resources>
+ <resource>
+ <directory>src/main/resources</directory>
+ <filtering>true</filtering>
+ </resource>
+ </resources>
+
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Command.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Command.java
new file mode 100644
index 0000000..3d4cef6
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Command.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+/**
+ * A marker interface for replication group command.
+ */
+public interface Command {
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ElectionPriority.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ElectionPriority.java
new file mode 100644
index 0000000..e4a1002
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ElectionPriority.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+/**
+ * Election priority constants.
+ */
+public class ElectionPriority {
+ /**
+ * Priority -1 means this node has disabled the priority election function.
+ */
+ public static final int DISABLED = -1;
+
+ /**
+ * Priority 0 is a special value so that a node will never participate in election.
+ */
+ public static final int NOT_ELECTABLE = 0;
+
+ /**
+ * Priority 1 is a minimum value for priority election.
+ */
+ public static final int MIN_VALUE = 1;
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Peer.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Peer.java
new file mode 100644
index 0000000..d8b94d3
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/Peer.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+import org.apache.ignite.network.NetworkMember;
+
+/**
+ * A participant of a replication group.
+ */
+public final class Peer {
+ /**
+ * Network node.
+ */
+ private final NetworkMember node;
+
+ /**
+ * Peer's local priority value, if node don't support priority election,
+ * this value is {@link ElectionPriority#DISABLED}.
+ */
+ private final int priority;
+
+ /**
+ * @param peer Peer.
+ */
+ public Peer(Peer peer) {
+ this.node = peer.getNode();
+ this.priority = peer.getPriority();
+ }
+
+ /**
+ * @param node Node.
+ */
+ public Peer(NetworkMember node) {
+ this(node, ElectionPriority.DISABLED);
+ }
+
+ /**
+ * @param node Node.
+ * @param priority Election priority.
+ */
+ public Peer(NetworkMember node, int priority) {
+ this.node = node;
+ this.priority = priority;
+ }
+
+ /**
+ * @return Node.
+ */
+ public NetworkMember getNode() {
+ return this.node;
+ }
+
+ /**
+ * @return Election priority.
+ */
+ public int getPriority() {
+ return priority;
+ }
+
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Peer peer = (Peer) o;
+
+ if (priority != peer.priority) return false;
+ if (!node.equals(peer.node)) return false;
+
+ return true;
+ }
+
+ @Override public int hashCode() {
+ int result = node.hashCode();
+ result = 31 * result + priority;
+ return result;
+ }
+
+ @Override public String toString() {
+ return node.name() + ":" + priority;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftErrorCode.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftErrorCode.java
new file mode 100644
index 0000000..0346ed7
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/RaftErrorCode.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+/**
+ * Error codes for raft protocol.
+ */
+public enum RaftErrorCode {
+ /** */
+ SUCCESS(1000, "Successful"),
+
+ /** */
+ NO_LEADER(1001, "No leader found within a timeout"),
+
+ /** */
+ LEADER_CHANGED(1002, "A peer is no longer a leader");
+
+ /** */
+ private final int code;
+
+ /** */
+ private final String desc;
+
+ /**
+ * @param code The code.
+ * @param desc The desctiption.
+ */
+ RaftErrorCode(int code, String desc) {
+ this.code = code;
+ this.desc = desc;
+ }
+
+ /**
+ * @return The code.
+ */
+ public int code() {
+ return code;
+ }
+
+ /**
+ * @return The description.
+ */
+ public String description() {
+ return desc;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ReadCommand.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ReadCommand.java
new file mode 100644
index 0000000..95b5661
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/ReadCommand.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+/**
+ * A read command.
+ */
+public interface ReadCommand extends Command {
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/WriteCommand.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/WriteCommand.java
new file mode 100644
index 0000000..e84db9e
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/WriteCommand.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client;
+
+/**
+ * A write command.
+ */
+public interface WriteCommand extends Command {
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/exception/RaftException.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/exception/RaftException.java
new file mode 100644
index 0000000..45cbeb9
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/exception/RaftException.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.exception;
+
+import org.apache.ignite.raft.client.RaftErrorCode;
+
+/**
+ * A raft exception containing code and description.
+ */
+public class RaftException extends RuntimeException {
+ private final RaftErrorCode code;
+
+ /**
+ * @param errCode Error code.
+ */
+ public RaftException(RaftErrorCode errCode) {
+ this.code = errCode;
+ }
+
+ /**
+ * @return Error code.
+ */
+ public RaftErrorCode errorCode() {
+ return code;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionRequest.java
new file mode 100644
index 0000000..104a7b6
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionRequest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.Command;
+
+/**
+ * Submit an action to a replication group.
+ */
+public interface ActionRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return Action's command.
+ */
+ Command command();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param cmd Action's command.
+ * @return The builder.
+ */
+ Builder command(Command cmd);
+
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ ActionRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionResponse.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionResponse.java
new file mode 100644
index 0000000..d456b49
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ActionResponse.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+/**
+ * The result of an action.
+ */
+public interface ActionResponse<T> {
+ /**
+ * @return A result for this request, can be of any type.
+ */
+ T result();
+
+ /** */
+ public interface Builder<T> {
+ /**
+ * @param result A result for this request.
+ * @return The builder.
+ */
+ Builder result(T result);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ ActionResponse build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequest.java
new file mode 100644
index 0000000..2b70552
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddLearnersRequest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Add learners.
+ */
+public interface AddLearnersRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return List of learners.
+ */
+ List<Peer> learners();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param learners Learners.
+ * @return The builder.
+ */
+ Builder learners(List<Peer> learner);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ AddLearnersRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeersRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeersRequest.java
new file mode 100644
index 0000000..5b10d28
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/AddPeersRequest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Add peers.
+ */
+public interface AddPeersRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return Peers.
+ */
+ List<Peer> peers();
+
+ /** */
+ interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param peers Peers.
+ * @return The builder.
+ */
+ Builder peers(List<Peer> peers);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ AddPeersRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponse.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponse.java
new file mode 100644
index 0000000..d4e729a
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/ChangePeersResponse.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Change peers result.
+ */
+public interface ChangePeersResponse {
+ /**
+ * @return Old peers.
+ */
+ List<Peer> oldPeers();
+
+ /**
+ * @return New peers.
+ */
+ List<Peer> newPeers();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param oldPeers Old peers.
+ * @return The builder.
+ */
+ Builder oldPeers(List<Peer> oldPeers);
+
+ /**
+ * @param newPeers New peers.
+ * @return The builder.
+ */
+ Builder newPeers(List<Peer> newPeers);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ ChangePeersResponse build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequest.java
new file mode 100644
index 0000000..87d4d5b
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderRequest.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+/**
+ * Get leader.
+ */
+public interface GetLeaderRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ GetLeaderRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponse.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponse.java
new file mode 100644
index 0000000..30dcc1c
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetLeaderResponse.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * A current leader.
+ */
+public interface GetLeaderResponse {
+ /**
+ * @return The leader.
+ */
+ Peer leader();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param leader Leader
+ * @return The builder.
+ */
+ Builder leader(Peer leaderId);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ GetLeaderResponse build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequest.java
new file mode 100644
index 0000000..03942f6
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersRequest.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+/** Get peers. */
+public interface GetPeersRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return {@code True} to list only alive nodes.
+ */
+ boolean onlyAlive();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param onlyGetAlive {@code True} to list only alive nodes.
+ * @return The builder.
+ */
+ Builder onlyAlive(boolean onlyGetAlive);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ GetPeersRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponse.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponse.java
new file mode 100644
index 0000000..69810aa
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/GetPeersResponse.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ *
+ */
+public interface GetPeersResponse {
+ /**
+ * @return Current peers.
+ */
+ List<Peer> peers();
+
+ /**
+ * @return Current leaners.
+ */
+ List<Peer> learners();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param peers Current peers.
+ * @return The builder.
+ */
+ Builder peers(List<Peer> peers);
+
+ /**
+ * @param learners Current learners.
+ * @return The builder.
+ */
+ Builder learners(List<Peer> learners);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ GetPeersResponse build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftErrorResponse.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftErrorResponse.java
new file mode 100644
index 0000000..0c45d02
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RaftErrorResponse.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.RaftErrorCode;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Raft error response. Also used as a default response when errorCode == {@link RaftErrorCode#SUCCESS}
+ */
+public interface RaftErrorResponse {
+ /**
+ * @return Error code.
+ */
+ public RaftErrorCode errorCode();
+
+ /**
+ * @return The new leader if a current leader is obsolete or null if not applicable.
+ */
+ public @Nullable Peer newLeader();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param errorCode Error code.
+ * @return The builder.
+ */
+ Builder errorCode(RaftErrorCode errorCode);
+
+ /**
+ * @param newLeader New leader.
+ * @return The builder.
+ */
+ Builder newLeader(Peer newLeader);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ RaftErrorResponse build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequest.java
new file mode 100644
index 0000000..7e63218
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemoveLearnersRequest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Remove learners.
+ */
+public interface RemoveLearnersRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return Learners to remove.
+ */
+ List<Peer> learners();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param learners Learners to remove.
+ * @return The builder.
+ */
+ Builder learners(List<Peer> learners);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ RemoveLearnersRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeersRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeersRequest.java
new file mode 100644
index 0000000..ac94fb4
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/RemovePeersRequest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Remove peers.
+ */
+public interface RemovePeersRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return Peers to remove.
+ */
+ List<Peer> peers();
+
+ /** */
+ interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param peers Peers to remove.
+ * @return The builder.
+ */
+ Builder peers(List<Peer> peers);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ RemovePeersRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequest.java
new file mode 100644
index 0000000..b72e9a7
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/SnapshotRequest.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+/**
+ * Take a local snapshot on the peer.
+ */
+public interface SnapshotRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ SnapshotRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeadershipRequest.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeadershipRequest.java
new file mode 100644
index 0000000..e8da0fc
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/TransferLeadershipRequest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message;
+
+import org.apache.ignite.raft.client.Peer;
+
+/**
+ * Transfer a leadership to receiving peer.
+ */
+public interface TransferLeadershipRequest {
+ /**
+ * @return Group id.
+ */
+ String groupId();
+
+ /**
+ * @return New leader.
+ */
+ Peer newLeader();
+
+ /** */
+ public interface Builder {
+ /**
+ * @param groupId Group id.
+ * @return The builder.
+ */
+ Builder groupId(String groupId);
+
+ /**
+ * @param newLeader New leader.
+ * @return The builder.
+ */
+ Builder peer(Peer newLeader);
+
+ /**
+ * @return The complete message.
+ * @throws IllegalStateException If the message is not in valid state.
+ */
+ TransferLeadershipRequest build();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionRequestImpl.java
new file mode 100644
index 0000000..ca8e068
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionRequestImpl.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.Command;
+import org.apache.ignite.raft.client.message.ActionRequest;
+
+/** */
+class ActionRequestImpl<T> implements ActionRequest, ActionRequest.Builder {
+ /** */
+ private Command cmd;
+
+ /** */
+ private String groupId;
+
+ /** {@inheritDoc} */
+ @Override public Command command() {
+ return cmd;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder command(Command cmd) {
+ this.cmd = cmd;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public ActionRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionResponseImpl.java
new file mode 100644
index 0000000..2056804
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ActionResponseImpl.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.ActionResponse;
+
+/** */
+class ActionResponseImpl<T> implements ActionResponse<T>, ActionResponse.Builder<T> {
+ /** */
+ private T result;
+
+ /** {@inheritDoc} */
+ @Override public T result() {
+ return result;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder result(T result) {
+ this.result = result;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public ActionResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddLearnersRequestImpl.java
new file mode 100644
index 0000000..abf615b
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddLearnersRequestImpl.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.AddLearnersRequest;
+
+/** */
+public class AddLearnersRequestImpl implements AddLearnersRequest, AddLearnersRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private List<Peer> learners;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> learners() {
+ return learners;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder learners(List<Peer> learners) {
+ this.learners = learners;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public AddLearnersRequest build() {
+ if (learners == null || groupId == null)
+ throw new IllegalStateException();
+
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddPeersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddPeersRequestImpl.java
new file mode 100644
index 0000000..1de6c2b
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/AddPeersRequestImpl.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.AddPeersRequest;
+
+/** */
+class AddPeersRequestImpl implements AddPeersRequest, AddPeersRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private List<Peer> peers;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> peers() {
+ return peers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder peers(List<Peer> peers) {
+ this.peers = peers;
+
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public AddPeersRequest build() {
+ if (peers == null)
+ throw new IllegalArgumentException();
+
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ChangePeersResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ChangePeersResponseImpl.java
new file mode 100644
index 0000000..905a32b
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/ChangePeersResponseImpl.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.ChangePeersResponse;
+
+/** */
+class ChangePeersResponseImpl implements ChangePeersResponse, ChangePeersResponse.Builder {
+ /** */
+ private List<Peer> oldPeers;
+
+ /** */
+ private List<Peer> newPeers;
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> oldPeers() {
+ return oldPeers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> newPeers() {
+ return newPeers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder oldPeers(List<Peer> oldPeers) {
+ this.oldPeers = oldPeers;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder newPeers(List<Peer> newPeers) {
+ this.newPeers = newPeers;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public ChangePeersResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderRequestImpl.java
new file mode 100644
index 0000000..d7f16dc
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderRequestImpl.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.GetLeaderRequest;
+
+/** */
+public class GetLeaderRequestImpl implements GetLeaderRequest, GetLeaderRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetLeaderRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderResponseImpl.java
new file mode 100644
index 0000000..7b95e41
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetLeaderResponseImpl.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.GetLeaderResponse;
+
+/** */
+public class GetLeaderResponseImpl implements GetLeaderResponse, GetLeaderResponse.Builder {
+ /** */
+ private Peer leader;
+
+ /** {@inheritDoc} */
+ @Override public Peer leader() {
+ return leader;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder leader(Peer leader) {
+ this.leader = leader;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetLeaderResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersRequestImpl.java
new file mode 100644
index 0000000..f746434
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersRequestImpl.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.GetPeersRequest;
+
+/** */
+class GetPeersRequestImpl implements GetPeersRequest, GetPeersRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private boolean onlyAlive;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onlyAlive() {
+ return onlyAlive;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder onlyAlive(boolean onlyGetAlive) {
+ this.onlyAlive = onlyGetAlive;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetPeersRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersResponseImpl.java
new file mode 100644
index 0000000..2c40f87
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/GetPeersResponseImpl.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.GetPeersResponse;
+
+/** */
+class GetPeersResponseImpl implements GetPeersResponse, GetPeersResponse.Builder {
+ /** */
+ private List<Peer> peers;
+
+ /** */
+ private List<Peer> learners;
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> peers() {
+ return peers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> learners() {
+ return learners;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder peers(List<Peer> peers) {
+ this.peers = peers;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder learners(List<Peer> learners) {
+ this.learners = learners;
+
+ return this;
+ }
+
+ @Override public GetPeersResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactory.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactory.java
new file mode 100644
index 0000000..7512dec
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactory.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.AddLearnersRequest;
+import org.apache.ignite.raft.client.message.AddPeersRequest;
+import org.apache.ignite.raft.client.message.ChangePeersResponse;
+import org.apache.ignite.raft.client.message.GetLeaderRequest;
+import org.apache.ignite.raft.client.message.GetLeaderResponse;
+import org.apache.ignite.raft.client.message.GetPeersRequest;
+import org.apache.ignite.raft.client.message.GetPeersResponse;
+import org.apache.ignite.raft.client.message.RaftErrorResponse;
+import org.apache.ignite.raft.client.message.RemoveLearnersRequest;
+import org.apache.ignite.raft.client.message.RemovePeersRequest;
+import org.apache.ignite.raft.client.message.SnapshotRequest;
+import org.apache.ignite.raft.client.message.TransferLeadershipRequest;
+import org.apache.ignite.raft.client.message.ActionRequest;
+import org.apache.ignite.raft.client.message.ActionResponse;
+
+/**
+ * A factory for immutable replication group messages.
+ */
+public interface RaftClientMessageFactory {
+ /**
+ * @return The builder.
+ */
+ AddPeersRequest.Builder addPeersRequest();
+
+ /**
+ * @return The builder.
+ */
+ RemovePeersRequest.Builder removePeerRequest();
+
+ /**
+ * @return The builder.
+ */
+ SnapshotRequest.Builder snapshotRequest();
+
+ /**
+ * @return The builder.
+ */
+ TransferLeadershipRequest.Builder transferLeaderRequest();
+
+ /**
+ * @return The builder.
+ */
+ GetLeaderRequest.Builder getLeaderRequest();
+
+ /**
+ * @return The builder.
+ */
+ GetLeaderResponse.Builder getLeaderResponse();
+
+ /**
+ * @return The builder.
+ */
+ GetPeersRequest.Builder getPeersRequest();
+
+ /**
+ * @return The builder.
+ */
+ GetPeersResponse.Builder getPeersResponse();
+
+ /**
+ * @return The builder.
+ */
+ AddLearnersRequest.Builder addLearnersRequest();
+
+ /**
+ * @return The builder.
+ */
+ RemoveLearnersRequest.Builder removeLearnersRequest();
+
+ /**
+ * @return The builder.
+ */
+ ChangePeersResponse.Builder changePeersResponse();
+
+ /**
+ * @return The builder.
+ */
+ ActionRequest.Builder actionRequest();
+
+ /**
+ * @return The builder.
+ */
+ ActionResponse.Builder actionResponse();
+
+ /**
+ * @return The builder.
+ */
+ RaftErrorResponse.Builder raftErrorResponse();
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactoryImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactoryImpl.java
new file mode 100644
index 0000000..ce52677
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftClientMessageFactoryImpl.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.AddLearnersRequest;
+import org.apache.ignite.raft.client.message.AddPeersRequest;
+import org.apache.ignite.raft.client.message.ChangePeersResponse;
+import org.apache.ignite.raft.client.message.GetLeaderRequest;
+import org.apache.ignite.raft.client.message.GetLeaderResponse;
+import org.apache.ignite.raft.client.message.GetPeersRequest;
+import org.apache.ignite.raft.client.message.GetPeersResponse;
+import org.apache.ignite.raft.client.message.RaftErrorResponse;
+import org.apache.ignite.raft.client.message.RemoveLearnersRequest;
+import org.apache.ignite.raft.client.message.RemovePeersRequest;
+import org.apache.ignite.raft.client.message.SnapshotRequest;
+import org.apache.ignite.raft.client.message.TransferLeadershipRequest;
+import org.apache.ignite.raft.client.message.ActionRequest;
+import org.apache.ignite.raft.client.message.ActionResponse;
+
+/**
+ * The default implementation.
+ */
+public class RaftClientMessageFactoryImpl implements RaftClientMessageFactory {
+ /** {@inheritDoc} */
+ @Override public AddPeersRequest.Builder addPeersRequest() {
+ return new AddPeersRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public ChangePeersResponse.Builder changePeersResponse() {
+ return new ChangePeersResponseImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RemovePeersRequest.Builder removePeerRequest() {
+ return new RemovePeersRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public SnapshotRequest.Builder snapshotRequest() {
+ return new SnapshotRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public TransferLeadershipRequest.Builder transferLeaderRequest() {
+ return new TransferLeadershipRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetLeaderRequest.Builder getLeaderRequest() {
+ return new GetLeaderRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetLeaderResponse.Builder getLeaderResponse() {
+ return new GetLeaderResponseImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetPeersRequest.Builder getPeersRequest() {
+ return new GetPeersRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GetPeersResponse.Builder getPeersResponse() {
+ return new GetPeersResponseImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public AddLearnersRequest.Builder addLearnersRequest() {
+ return new AddLearnersRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RemoveLearnersRequest.Builder removeLearnersRequest() {
+ return new RemoveLearnersRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public ActionRequest.Builder actionRequest() {
+ return new ActionRequestImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public ActionResponse.Builder actionResponse() {
+ return new ActionResponseImpl();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RaftErrorResponse.Builder raftErrorResponse() {
+ return new RaftErrorResponseImpl();
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftErrorResponseImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftErrorResponseImpl.java
new file mode 100644
index 0000000..5ed91e3
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RaftErrorResponseImpl.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.RaftErrorCode;
+import org.apache.ignite.raft.client.message.RaftErrorResponse;
+
+public class RaftErrorResponseImpl implements RaftErrorResponse, RaftErrorResponse.Builder {
+ /** */
+ private RaftErrorCode errorCode;
+
+ /** */
+ private Peer newLeader;
+
+ /** {@inheritDoc} */
+ @Override public RaftErrorCode errorCode() {
+ return errorCode;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Peer newLeader() {
+ return newLeader;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder errorCode(RaftErrorCode errorCode) {
+ this.errorCode = errorCode;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder newLeader(Peer newLeader) {
+ this.newLeader = newLeader;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public RaftErrorResponse build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemoveLearnersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemoveLearnersRequestImpl.java
new file mode 100644
index 0000000..b3e690a
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemoveLearnersRequestImpl.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.RemoveLearnersRequest;
+
+/** */
+class RemoveLearnersRequestImpl implements RemoveLearnersRequest, RemoveLearnersRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private List<Peer> learners = new ArrayList<>();
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> learners() {
+ return learners;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder learners(List<Peer> learners) {
+ this.learners = learners;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public RemoveLearnersRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemovePeersRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemovePeersRequestImpl.java
new file mode 100644
index 0000000..5fd5572
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/RemovePeersRequestImpl.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import java.util.List;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.RemovePeersRequest;
+
+/** */
+class RemovePeersRequestImpl implements RemovePeersRequest, RemovePeersRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private List<Peer> peers;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> peers() {
+ return peers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder peers(List<Peer> peers) {
+ this.peers = peers;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public RemovePeersRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/SnapshotRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/SnapshotRequestImpl.java
new file mode 100644
index 0000000..bc1725a
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/SnapshotRequestImpl.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.message.SnapshotRequest;
+
+/** */
+class SnapshotRequestImpl implements SnapshotRequest, SnapshotRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public SnapshotRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/TransferLeadershipRequestImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/TransferLeadershipRequestImpl.java
new file mode 100644
index 0000000..4286fff
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/message/impl/TransferLeadershipRequestImpl.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.message.impl;
+
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.message.TransferLeadershipRequest;
+
+/** */
+class TransferLeadershipRequestImpl implements TransferLeadershipRequest, TransferLeadershipRequest.Builder {
+ /** */
+ private String groupId;
+
+ /** */
+ private Peer newLeader;
+
+ /** {@inheritDoc} */
+ @Override public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * @return New leader.
+ */
+ @Override public Peer newLeader() {
+ return newLeader;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder groupId(String groupId) {
+ this.groupId = groupId;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Builder peer(Peer newLeader) {
+ this.newLeader = newLeader;
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public TransferLeadershipRequest build() {
+ return this;
+ }
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupCommandListener.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupCommandListener.java
new file mode 100644
index 0000000..d22c334
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupCommandListener.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.service;
+
+import java.util.Iterator;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+
+/**
+ * A listener for replication group commands.
+ */
+public interface RaftGroupCommandListener {
+ /**
+ * @param iterator Read command iterator.
+ */
+ void onRead(Iterator<ReadCommand> iterator);
+
+ /**
+ * @param iterator Write command iterator.
+ */
+ void onWrite(Iterator<WriteCommand> iterator);
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupService.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupService.java
new file mode 100644
index 0000000..4c3d957
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/RaftGroupService.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.service;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.raft.client.Command;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A service providing operations on a replication group.
+ * <p>
+ * Most of operations require a known group leader. The group leader can be refreshed at any time by calling
+ * {@link #refreshLeader()} method, otherwise it will happen automatically on a first call.
+ * <p>
+ * If a leader has been changed while the operation in progress, the operation will be transparently retried until
+ * timeout is reached. The current leader will be refreshed automatically (maybe several times) in the process.
+ * <p>
+ * Each asynchronous method (returning a future) uses a default timeout to finish, see {@link #timeout()}.
+ * If a result is not available within the timeout, the future will be completed with a {@link TimeoutException}
+ * <p>
+ * If an error is occured during operation execution, the future will be completed with the corresponding
+ * IgniteException having an error code and a related message.
+ * <p>
+ * Async operations provided by the service are not cancellable.
+ */
+public interface RaftGroupService {
+ /**
+ * @return Group id.
+ */
+ @NotNull String groupId();
+
+ /**
+ * @return Default timeout for the operations in milliseconds.
+ */
+ long timeout();
+
+ /**
+ * Changes default timeout value for all subsequent operations.
+ *
+ * @param newTimeout New timeout value.
+ */
+ void timeout(long newTimeout);
+
+ /**
+ * @return Current leader id or {@code null} if it has not been yet initialized.
+ */
+ @Nullable Peer leader();
+
+ /**
+ * @return A list of voting peers or {@code null} if it has not been yet initialized. The order is corresponding
+ * to the time of joining to the replication group.
+ */
+ @Nullable List<Peer> peers();
+
+ /**
+ * @return A list of leaners or {@code null} if it has not been yet initialized. The order is corresponding
+ * to the time of joining to the replication group.
+ */
+ @Nullable List<Peer> learners();
+
+ /**
+ * Refreshes a replication group leader.
+ * <p>
+ * After the future completion the method {@link #leader()}
+ * can be used to retrieve a current group leader.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @return A future.
+ */
+ CompletableFuture<Void> refreshLeader();
+
+ /**
+ * Refreshes replication group members.
+ * <p>
+ * After the future completion methods like {@link #peers()} and {@link #learners()}
+ * can be used to retrieve current members of a group.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param onlyAlive {@code True} to exclude dead nodes.
+ * @return A future.
+ */
+ CompletableFuture<Void> refreshMembers(boolean onlyAlive);
+
+ /**
+ * Adds a voting peers to the replication group.
+ * <p>
+ * After the future completion methods like {@link #peers()} and {@link #learners()}
+ * can be used to retrieve current members of a group.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param peers Peers.
+ * @return A future.
+ */
+ CompletableFuture<Void> addPeers(List<Peer> peers);
+
+ /**
+ * Removes peers from the replication group.
+ * <p>
+ * After the future completion methods like {@link #peers()} and {@link #learners()}
+ * can be used to retrieve current members of a group.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param peers Peers.
+ * @return A future.
+ */
+ CompletableFuture<Void> removePeers(List<Peer> peers);
+
+ /**
+ * Adds learners (non-voting members).
+ * <p>
+ * After the future completion methods like {@link #peers()} and {@link #learners()}
+ * can be used to retrieve current members of a group.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param learners List of learners.
+ * @return A future.
+ */
+ CompletableFuture<Void> addLearners(List<Peer> learners);
+
+ /**
+ * Removes learners.
+ * <p>
+ * After the future completion methods like {@link #peers()} and {@link #learners()}
+ * can be used to retrieve current members of a group.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param learners List of learners.
+ * @return A future.
+ */
+ CompletableFuture<Void> removeLearners(List<Peer> learners);
+
+ /**
+ * Takes a state machine snapshot on a given group peer.
+ *
+ * @param peer Peer.
+ * @return A future.
+ */
+ CompletableFuture<Void> snapshot(Peer peer);
+
+ /**
+ * Transfers leadership to other peer.
+ * <p>
+ * This operation is executed on a group leader.
+ *
+ * @param newLeader New leader.
+ * @return A future.
+ */
+ CompletableFuture<Void> transferLeadership(Peer newLeader);
+
+ /**
+ * Runs a command on a replication group leader.
+ * <p>
+ * Read commands always see up to date data.
+ *
+ * @param cmd The command.
+ * @return A future with the execution result.
+ */
+ <R> CompletableFuture<R> run(Command cmd);
+
+ /**
+ * Runs a read command on a given peer.
+ * <p>
+ * Read commands can see stale data (in the past).
+ *
+ * @param peer Peer id.
+ * @param cmd The command.
+ * @return A future with the execution result.
+ */
+ <R> CompletableFuture<R> run(Peer peer, ReadCommand cmd);
+}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupServiceImpl.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupServiceImpl.java
new file mode 100644
index 0000000..4e0df31
--- /dev/null
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/impl/RaftGroupServiceImpl.java
@@ -0,0 +1,389 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.service.impl;
+
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
+import org.apache.ignite.lang.LogWrapper;
+import org.apache.ignite.network.NetworkCluster;
+import org.apache.ignite.network.NetworkMember;
+import org.apache.ignite.raft.client.Command;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.exception.RaftException;
+import org.apache.ignite.raft.client.message.AddLearnersRequest;
+import org.apache.ignite.raft.client.message.AddPeersRequest;
+import org.apache.ignite.raft.client.message.ChangePeersResponse;
+import org.apache.ignite.raft.client.message.GetLeaderResponse;
+import org.apache.ignite.raft.client.message.RaftErrorResponse;
+import org.apache.ignite.raft.client.message.GetLeaderRequest;
+import org.apache.ignite.raft.client.message.GetPeersRequest;
+import org.apache.ignite.raft.client.message.GetPeersResponse;
+import org.apache.ignite.raft.client.message.ActionRequest;
+import org.apache.ignite.raft.client.message.ActionResponse;
+import org.apache.ignite.raft.client.message.RemoveLearnersRequest;
+import org.apache.ignite.raft.client.message.RemovePeersRequest;
+import org.apache.ignite.raft.client.message.SnapshotRequest;
+import org.apache.ignite.raft.client.message.TransferLeadershipRequest;
+import org.apache.ignite.raft.client.message.impl.RaftClientMessageFactory;
+import org.apache.ignite.raft.client.service.RaftGroupService;
+import org.jetbrains.annotations.NotNull;
+
+import static java.lang.System.currentTimeMillis;
+import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.ThreadLocalRandom.current;
+import static org.apache.ignite.raft.client.RaftErrorCode.LEADER_CHANGED;
+import static org.apache.ignite.raft.client.RaftErrorCode.NO_LEADER;
+import static org.apache.ignite.raft.client.RaftErrorCode.SUCCESS;
+
+/**
+ * The implementation of {@link RaftGroupService}
+ */
+public class RaftGroupServiceImpl implements RaftGroupService {
+ /** */
+ private static LogWrapper LOG = new LogWrapper(RaftGroupServiceImpl.class);
+
+ /** */
+ private volatile int timeout;
+
+ /** */
+ private final String groupId;
+
+ /** */
+ private final RaftClientMessageFactory factory;
+
+ /** */
+ private volatile Peer leader;
+
+ /** */
+ private volatile List<Peer> peers;
+
+ /** */
+ private volatile List<Peer> learners;
+
+ /** */
+ private final NetworkCluster cluster;
+
+ /** */
+ private final long retryDelay;
+
+ /** */
+ private final Timer timer;
+
+ /**
+ * @param groupId Group id.
+ * @param cluster A cluster.
+ * @param factory A message factory.
+ * @param timeout Request timeout.
+ * @param peers Initial group configuration.
+ * @param refreshLeader {@code True} to synchronously refresh leader on service creation.
+ * @param retryDelay Retry delay.
+ * @param timer Timer for scheduled execution.
+ */
+ public RaftGroupServiceImpl(
+ String groupId,
+ NetworkCluster cluster,
+ RaftClientMessageFactory factory,
+ int timeout,
+ List<Peer> peers,
+ boolean refreshLeader,
+ long retryDelay,
+ Timer timer
+ ) {
+ this.cluster = requireNonNull(cluster);
+ this.peers = requireNonNull(peers);
+ this.factory = factory;
+ this.timeout = timeout;
+ this.groupId = groupId;
+ this.retryDelay = retryDelay;
+ this.timer = requireNonNull(timer);
+
+ if (refreshLeader) {
+ try {
+ refreshLeader().get();
+ }
+ catch (Exception e) {
+ LOG.error("Failed to refresh a leader", e);
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public @NotNull String groupId() {
+ return groupId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long timeout() {
+ return timeout;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void timeout(long newTimeout) {
+ this.timeout = timeout;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Peer leader() {
+ return leader;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> peers() {
+ return peers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<Peer> learners() {
+ return learners;
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> refreshLeader() {
+ GetLeaderRequest req = factory.getLeaderRequest().groupId(groupId).build();
+
+ CompletableFuture<GetLeaderResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(randomNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ leader = resp.leader();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> refreshMembers(boolean onlyAlive) {
+ GetPeersRequest req = factory.getPeersRequest().onlyAlive(onlyAlive).groupId(groupId).build();
+
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> refreshMembers(onlyAlive));
+
+ CompletableFuture<GetPeersResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ peers = resp.peers();
+ learners = resp.learners();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> addPeers(List<Peer> peers) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> addPeers(peers));
+
+ AddPeersRequest req = factory.addPeersRequest().groupId(groupId).peers(peers).build();
+
+ CompletableFuture<ChangePeersResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ this.peers = resp.newPeers();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> removePeers(List<Peer> peers) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> removePeers(peers));
+
+ RemovePeersRequest req = factory.removePeerRequest().groupId(groupId).peers(peers).build();
+
+ CompletableFuture<ChangePeersResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ this.peers = resp.newPeers();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> addLearners(List<Peer> learners) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> addLearners(learners));
+
+ AddLearnersRequest req = factory.addLearnersRequest().groupId(groupId).learners(learners).build();
+
+ CompletableFuture<ChangePeersResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ this.learners = resp.newPeers();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> removeLearners(List<Peer> learners) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> removeLearners(learners));
+
+ RemoveLearnersRequest req = factory.removeLearnersRequest().groupId(groupId).learners(learners).build();
+
+ CompletableFuture<ChangePeersResponse> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> {
+ this.learners = resp.newPeers();
+
+ return null;
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> snapshot(Peer peer) {
+ SnapshotRequest req = factory.snapshotRequest().groupId(groupId).build();
+
+ CompletableFuture<?> fut = cluster.sendWithResponse(peer.getNode(), req, timeout);
+
+ return fut.thenApply(resp -> null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> transferLeadership(Peer newLeader) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> transferLeadership(newLeader));
+
+ TransferLeadershipRequest req = factory.transferLeaderRequest().groupId(groupId).peer(newLeader).build();
+
+ CompletableFuture<?> fut = cluster.sendWithResponse(newLeader.getNode(), req, timeout);
+
+ return fut.thenApply(resp -> null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <R> CompletableFuture<R> run(Command cmd) {
+ Peer leader = this.leader;
+
+ if (leader == null)
+ return refreshLeader().thenCompose(res -> run(cmd));
+
+ ActionRequest req = factory.actionRequest().command(cmd).groupId(groupId).build();
+
+ CompletableFuture<ActionResponse<R>> fut = new CompletableFuture<>();
+
+ sendWithRetry(leader.getNode(), req, currentTimeMillis() + timeout, fut);
+
+ return fut.thenApply(resp -> resp.result());
+ }
+
+ /** {@inheritDoc} */
+ @Override public <R> CompletableFuture<R> run(Peer peer, ReadCommand cmd) {
+ ActionRequest req = factory.actionRequest().command(cmd).groupId(groupId).build();
+
+ CompletableFuture fut = cluster.sendWithResponse(peer.getNode(), req, timeout);
+
+ return fut.thenApply(resp -> ((ActionResponse) resp).result());
+ }
+
+ /**
+ * Retries request until success or time is run out.
+ *
+ * @param req Request.
+ * @param stopTime Stop time.
+ * @param <R> Return value.
+ * @return A future.
+ */
+ private <R> void sendWithRetry(NetworkMember node, Object req, long stopTime, CompletableFuture<R> fut) {
+ if (currentTimeMillis() >= stopTime) {
+ fut.completeExceptionally(new TimeoutException());
+
+ return;
+ }
+
+ CompletableFuture fut0 = cluster.sendWithResponse(node, req, timeout);
+
+ fut0.whenComplete(new BiConsumer<Object, Throwable>() {
+ @Override public void accept(Object resp, Throwable err) {
+ if (err != null)
+ fut.completeExceptionally(err);
+ else {
+ if (resp instanceof RaftErrorResponse) {
+ RaftErrorResponse resp0 = (RaftErrorResponse) resp;
+
+ if (resp0.errorCode().equals(NO_LEADER)) {
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ sendWithRetry(randomNode(), req, stopTime, fut);
+ }
+ }, retryDelay);
+ }
+ else if (resp0.errorCode().equals(LEADER_CHANGED)) {
+ leader = resp0.newLeader(); // Update a leader.
+
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ sendWithRetry(resp0.newLeader().getNode(), req, stopTime, fut);
+ }
+ }, retryDelay);
+ }
+ else if (resp0.errorCode().equals(SUCCESS)) { // Handle default response.
+ fut.complete(null);
+ }
+ else
+ fut.completeExceptionally(new RaftException(resp0.errorCode()));
+ }
+ else
+ fut.complete((R) resp);
+ }
+ }
+ });
+ }
+
+ /**
+ * @return Random node.
+ */
+ private NetworkMember randomNode() {
+ List<Peer> peers0 = peers;
+
+ if (peers0 == null || peers0.isEmpty())
+ return null;
+
+ return peers0.get(current().nextInt(peers0.size())).getNode();
+ }
+}
diff --git a/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupServiceTest.java b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupServiceTest.java
new file mode 100644
index 0000000..038b43a
--- /dev/null
+++ b/modules/raft-client/src/test/java/org/apache/ignite/raft/client/service/RaftGroupServiceTest.java
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.raft.client.service;
+
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.lang.LogWrapper;
+import org.apache.ignite.network.NetworkCluster;
+import org.apache.ignite.network.NetworkMember;
+import org.apache.ignite.raft.client.Peer;
+import org.apache.ignite.raft.client.RaftErrorCode;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.message.ActionRequest;
+import org.apache.ignite.raft.client.message.GetLeaderRequest;
+import org.apache.ignite.raft.client.message.impl.RaftClientMessageFactory;
+import org.apache.ignite.raft.client.message.impl.RaftClientMessageFactoryImpl;
+import org.apache.ignite.raft.client.service.impl.RaftGroupServiceImpl;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+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.List.of;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+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.argThat;
+
+/**
+ * Test methods of raft group service.
+ */
+@ExtendWith(MockitoExtension.class)
+public class RaftGroupServiceTest {
+ /** */
+ private static LogWrapper LOG = new LogWrapper(RaftGroupServiceTest.class);
+
+ /** */
+ private static List<Peer> NODES = of(new Peer(new NetworkMember("node1")), new Peer(new NetworkMember("node2")),
+ new Peer(new NetworkMember("node3")));
+
+ /** */
+ private static RaftClientMessageFactory FACTORY = new RaftClientMessageFactoryImpl();
+
+ /** */
+ private volatile Peer leader = NODES.get(0);
+
+ /** Call timeout. */
+ private static final int TIMEOUT = 1000;
+
+ /** Retry delay. */
+ private static final int DELAY = 200;
+
+ /** Cluster. */
+ @Mock
+ private NetworkCluster cluster;
+
+ /**
+ * @param testInfo Test info.
+ */
+ @BeforeEach
+ void before(TestInfo testInfo) {
+ LOG.info(">>>> Starting test " + testInfo.getTestMethod().orElseThrow().getName());
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testRefreshLeaderStable() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ assertNull(service.leader());
+
+ service.refreshLeader().get();
+
+ assertEquals(leader, service.leader());
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testRefreshLeaderNotElected() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+
+ // Simulate running elections.
+ leader = null;
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ assertNull(service.leader());
+
+ try {
+ service.refreshLeader().get();
+
+ fail("Should fail");
+ }
+ catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testRefreshLeaderElectedAfterDelay() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+
+ // Simulate running elections.
+ leader = null;
+
+ Timer timer = new Timer();
+
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ leader = NODES.get(0);
+ }
+ }, 500);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, timer);
+
+ assertNull(service.leader());
+
+ service.refreshLeader().get();
+
+ assertEquals(NODES.get(0), service.leader());
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testRefreshLeaderWithTimeout() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, true);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ try {
+ service.refreshLeader().get();
+
+ fail();
+ }
+ catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestLeaderElected() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ service.refreshLeader().get();
+
+ TestResponse resp = service.<TestResponse>run(new TestCommand()).get();
+
+ assertNotNull(resp);
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestLazyInitLeader() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ assertNull(service.leader());
+
+ TestResponse resp = service.<TestResponse>run(new TestCommand()).get();
+
+ assertNotNull(resp);
+
+ assertEquals(leader, service.leader());
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestWithTimeout() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, true);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, false, DELAY, new Timer());
+
+ try {
+ service.run(new TestCommand()).get();
+
+ fail();
+ }
+ catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestLeaderNotElected() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, true, DELAY, new Timer());
+
+ Peer leader = this.leader;
+
+ assertEquals(leader, service.leader());
+
+ this.leader = null;
+
+ assertEquals(leader, service.leader());
+
+ try {
+ service.run(new TestCommand()).get();
+
+ fail("Expecting timeout");
+ }
+ catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ }
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestLeaderElectedAfterDelay() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, true, DELAY, new Timer());
+
+ Peer leader = this.leader;
+
+ assertEquals(leader, service.leader());
+
+ this.leader = null;
+
+ assertEquals(leader, service.leader());
+
+ Timer timer = new Timer();
+
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ RaftGroupServiceTest.this.leader = NODES.get(0);
+ }
+ }, 500);
+
+ TestResponse resp = service.<TestResponse>run(new TestCommand()).get();
+
+ assertNotNull(resp);
+
+ assertEquals(NODES.get(0), service.leader());
+ }
+
+ /**
+ * @throws Exception
+ */
+ @Test
+ public void testUserRequestLeaderChanged() throws Exception {
+ String groupId = "test";
+
+ mockLeaderRequest(cluster, false);
+ mockUserInput(cluster, false);
+
+ RaftGroupService service =
+ new RaftGroupServiceImpl(groupId, cluster, FACTORY, TIMEOUT, NODES, true, DELAY, new Timer());
+
+ Peer leader = this.leader;
+
+ assertEquals(leader, service.leader());
+
+ Peer newLeader = NODES.get(1);
+
+ this.leader = newLeader;
+
+ assertEquals(leader, service.leader());
+ assertNotEquals(leader, newLeader);
+
+ // Runs the command on an old leader. It should respond with leader changed error, when transparently retry.
+ TestResponse resp = service.<TestResponse>run(new TestCommand()).get();
+
+ assertNotNull(resp);
+
+ assertEquals(newLeader, service.leader());
+ }
+
+ /**
+ * @param cluster The cluster.
+ * @param simulateTimeout {@code True} to simulate request timeout.
+ */
+ private void mockUserInput(NetworkCluster cluster, boolean simulateTimeout) {
+ Mockito.doAnswer(new Answer() {
+ @Override public Object answer(InvocationOnMock invocation) throws Throwable {
+ NetworkMember target = invocation.getArgument(0);
+
+ if (simulateTimeout)
+ return failedFuture(new TimeoutException());
+
+ Object resp;
+
+ if (leader == null)
+ resp = FACTORY.raftErrorResponse().errorCode(RaftErrorCode.NO_LEADER).build();
+ else if (target != leader.getNode())
+ resp = FACTORY.raftErrorResponse().errorCode(RaftErrorCode.LEADER_CHANGED).newLeader(leader).build();
+ else
+ resp = FACTORY.actionResponse().result(new TestResponse()).build();
+
+ return completedFuture(resp);
+ }
+ }).when(cluster).sendWithResponse(any(), argThat(new ArgumentMatcher<ActionRequest>() {
+ @Override public boolean matches(ActionRequest arg) {
+ return arg.command() instanceof TestCommand;
+ }
+ }), anyLong());
+ }
+
+ /**
+ * @param cluster The cluster.
+ * @param simulateTimeout {@code True} to simulate request timeout.
+ */
+ private void mockLeaderRequest(NetworkCluster cluster, boolean simulateTimeout) {
+ Mockito.doAnswer(new Answer() {
+ @Override public Object answer(InvocationOnMock invocation) throws Throwable {
+ if (simulateTimeout)
+ return failedFuture(new TimeoutException());
+
+ Object resp;
+
+ Peer leader0 = leader;
+
+ if (leader0 == null) {
+ resp = FACTORY.raftErrorResponse().errorCode(RaftErrorCode.NO_LEADER).build();
+ }
+ else {
+ resp = FACTORY.getLeaderResponse().leader(leader0).build();
+ }
+
+ return completedFuture(resp);
+ }
+ }).when(cluster).sendWithResponse(any(), any(GetLeaderRequest.class), anyLong());
+ }
+
+ /** */
+ private static class TestCommand implements WriteCommand {
+ }
+
+ /** */
+ private static class TestResponse {
+ }
+}
diff --git a/pom.xml b/pom.xml
index 1c0872b..75141f2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -42,6 +42,7 @@
<module>modules/configuration-annotation-processor</module>
<module>modules/core</module>
<module>modules/network</module>
+ <module>modules/raft-client</module>
<module>modules/rest</module>
<module>modules/runner</module>
<module>modules/schema</module>