vldpyatkov commented on code in PR #5221:
URL: https://github.com/apache/ignite-3/pull/5221#discussion_r1987284099
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/core/DefaultRaftClientService.java:
##########
@@ -88,19 +93,78 @@ public Future<Message> requestVote(final PeerId peerId,
final RequestVoteRequest
@Override
public Future<Message> appendEntries(final PeerId peerId, final
AppendEntriesRequest request,
- final int timeoutMs, final RpcResponseClosure<AppendEntriesResponse>
done) {
+ final int timeoutMs, final
RpcResponseClosure<AppendEntriesResponse> done) {
// Assign an executor in round-robin fasion.
final Executor executor =
this.appendEntriesExecutorMap.computeIfAbsent(peerId,
- k -> nodeOptions.getStripedExecutor().next());
+ k -> nodeOptions.getStripedExecutor().next());
if (connect(peerId)) { // Replicator should be started asynchronously
by node joined event.
+ if (isHeartbeatRequest(request)) {
+ return sendHeartbeat(peerId, request, timeoutMs, done,
executor);
+ }
+
return invokeWithDone(peerId, request, done, timeoutMs, executor);
}
return onConnectionFail(executor, request, done, peerId);
}
+ /**
+ * Accumulates heartbeat messages to send them into the batch request.
+ *
+ * @param peerId Remote peer id.
+ * @param request Request.
+ * @param timeoutMs Timeout.
+ * @param done Done callback.
+ * @param executor Executor where the done callback is executed.
+ * @return A future with response.
+ */
+ private Future<Message> sendHeartbeat(
+ PeerId peerId,
+ AppendEntriesRequest request,
+ int timeoutMs,
+ RpcResponseClosure<AppendEntriesResponse> done,
+ Executor executor
+ ) {
+ NodeManager nodeManager = this.nodeOptions.getNodeManager();
+
+ return invokeWithDone(
+ peerId,
+ request,
+ null,
+ done,
+ timeoutMs,
+ executor,
+ (peerId1, request1, ctx, callback, timeoutMs1) ->
+ nodeManager.enqueue(peerId, (Message)
request1).whenComplete((res, err) -> {
+ if (err instanceof ExecutionException) {
+ err = new RemotingException(err);
+ } else if (err instanceof TimeoutException) //
Translate timeout exception.
+ {
+ err = new InvokeTimeoutException();
+ }
+
+ Throwable finalErr = err;
+
+ // Avoid deadlocks if a closure has completed in
the same thread.
+ Utils.runInThread(callback.executor(), () ->
callback.complete(res, finalErr));
+ })
+ );
+ }
+
+ /**
+ * Determines whether it is a heartbeat request.
+ *
+ * @param request Append entries request.
+ * @return True if that request is heartbeat or false otherwise.
+ */
+ private static boolean isHeartbeatRequest(final AppendEntriesRequest
request) {
Review Comment:
The same as AppendEntriesRequestProcessor#isHeartbeatRequest
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
