class RpcForwarder {

    private static final Logger LOG = Log.logger(RpcForwarder.class);

    private final PeerId nodeId;
    private final RaftClientService rpcClient;

    public RpcForwarder(Node node) {
        this.nodeId = node.getNodeId().getPeerId();
        this.rpcClient = ((NodeImpl) node).getRpcService();
        E.checkNotNull(this.rpcClient, "rpc client");
    }

    public void forwardToLeader(PeerId leaderId, StoreCommand command,
                                RaftStoreClosure future) {<FILL_FUNCTION_BODY>}

    public <T extends Message> RaftClosure<T> forwardToLeader(PeerId leaderId,
                                                              Message request) {
        E.checkNotNull(leaderId, "leader id");
        E.checkState(!leaderId.equals(this.nodeId),
                     "Invalid state: current node is the leader, there is " +
                     "no need to forward the request");
        LOG.debug("The node '{}' forward request to leader '{}'",
                  this.nodeId, leaderId);

        RaftClosure<T> future = new RaftClosure<>();
        RpcResponseClosure<T> responseDone = new RpcResponseClosure<T>() {
            @Override
            public void setResponse(T response) {
                FieldDescriptor fd = response.getDescriptorForType()
                                             .findFieldByName("common");
                Object object = response.getField(fd);
                E.checkState(object instanceof CommonResponse,
                             "The common field must be instance of " +
                             "CommonResponse, actual is '%s'",
                             object != null ? object.getClass() : null);
                CommonResponse commonResponse = (CommonResponse) object;
                if (commonResponse.getStatus()) {
                    future.complete(Status.OK(), () -> response);
                } else {
                    Status status = new Status(RaftError.UNKNOWN,
                                               "fowared request failed");
                    BackendException e = new BackendException(
                            "Current node isn't leader, leader " +
                            "is [%s], failed to forward request " +
                            "to leader: %s",
                            leaderId, commonResponse.getMessage());
                    future.failure(status, e);
                }
            }

            @Override
            public void run(Status status) {
                future.run(status);
            }
        };
        this.waitRpc(leaderId.getEndpoint(), request, responseDone);
        return future;
    }

    private <T extends Message> void waitRpc(Endpoint endpoint, Message request,
                                             RpcResponseClosure<T> done) {
        E.checkNotNull(endpoint, "leader endpoint");
        try {
            this.rpcClient.invokeWithDone(endpoint, request, done,
                                          RaftContext.WAIT_RPC_TIMEOUT)
                          .get();
        } catch (InterruptedException e) {
            throw new BackendException("Invoke rpc request was interrupted, " +
                                       "please try again later", e);
        } catch (ExecutionException e) {
            throw new BackendException("Failed to invoke rpc request", e);
        }
    }
}

        E.checkNotNull(leaderId, "leader id");
        E.checkState(!leaderId.equals(this.nodeId),
                     "Invalid state: current node is the leader, there is " +
                     "no need to forward the request");
        LOG.debug("The node {} forward request to leader {}",
                  this.nodeId, leaderId);

        StoreCommandRequest.Builder builder = StoreCommandRequest.newBuilder();
        builder.setType(command.type());
        builder.setAction(command.action());
        builder.setData(ZeroByteStringHelper.wrap(command.data()));
        StoreCommandRequest request = builder.build();

        RpcResponseClosure<StoreCommandResponse> responseClosure;
        responseClosure = new RpcResponseClosure<StoreCommandResponse>() {
            @Override
            public void setResponse(StoreCommandResponse response) {
                if (response.getStatus()) {
                    LOG.debug("StoreCommandResponse status ok");
                    // This code forwards the request to the Raft leader and considers the
                    // operation successful
                    // if it's forwarded successfully. It returns a RaftClosure because the calling
                    // logic expects a RaftClosure result. Specifically, if the current instance
                    // is the Raft leader,
                    // it executes the corresponding logic locally and notifies the calling logic
                    // asynchronously
                    // via RaftClosure. Therefore, the result is returned as a RaftClosure here.
                    RaftClosure<Status> supplierFuture = new RaftClosure<>();
                    supplierFuture.complete(Status.OK());
                    future.complete(Status.OK(), () -> supplierFuture);
                } else {
                    LOG.debug("StoreCommandResponse status error");
                    Status status = new Status(RaftError.UNKNOWN,
                                               "fowared request failed");
                    BackendException e = new BackendException(
                            "Current node isn't leader, leader " +
                            "is [%s], failed to forward request " +
                            "to leader: %s",
                            leaderId, response.getMessage());
                    future.failure(status, e);
                }
            }

            @Override
            public void run(Status status) {
                future.run(status);
            }
        };
        this.waitRpc(leaderId.getEndpoint(), request, responseClosure);
    
<no_super_class>