You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by lj...@apache.org on 2020/04/28 07:29:05 UTC
[incubator-ratis] branch master updated: RATIS-773. Fix checkstyle
violations in ratis-server. Contributed by Dinesh Chitlangia.
This is an automated email from the ASF dual-hosted git repository.
ljain pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ratis.git
The following commit(s) were added to refs/heads/master by this push:
new 14927c7 RATIS-773. Fix checkstyle violations in ratis-server. Contributed by Dinesh Chitlangia.
14927c7 is described below
commit 14927c7527afe13efb949065bbac0042ad82932f
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Tue Apr 28 12:46:41 2020 +0530
RATIS-773. Fix checkstyle violations in ratis-server. Contributed by Dinesh Chitlangia.
---
.../apache/ratis/grpc/server/GrpcLogAppender.java | 42 +++++++++---------
.../ratis/logservice/server/LogStateMachine.java | 6 +--
.../java/org/apache/ratis/server/RaftServer.java | 3 +-
.../org/apache/ratis/server/RaftServerRpc.java | 4 +-
.../apache/ratis/server/impl/LeaderElection.java | 9 ++--
.../org/apache/ratis/server/impl/LeaderState.java | 31 ++++++++------
.../org/apache/ratis/server/impl/LogAppender.java | 23 +++++++---
.../apache/ratis/server/impl/PendingRequests.java | 6 ++-
.../ratis/server/impl/RaftConfiguration.java | 4 +-
.../ratis/server/impl/RaftServerConstants.java | 12 ++++--
.../apache/ratis/server/impl/RaftServerImpl.java | 12 +++---
.../apache/ratis/server/impl/RaftServerProxy.java | 5 ++-
.../apache/ratis/server/impl/ServerImplUtils.java | 7 ++-
.../apache/ratis/server/impl/ServerProtoUtils.java | 2 +
.../org/apache/ratis/server/impl/ServerState.java | 12 +++---
.../ratis/server/impl/StateMachineUpdater.java | 6 +--
.../org/apache/ratis/server/raftlog/RaftLog.java | 3 +-
.../ratis/server/raftlog/memory/MemoryRaftLog.java | 20 ++++-----
.../ratis/server/raftlog/segmented/LogSegment.java | 9 +++-
.../raftlog/segmented/SegmentedRaftLogCache.java | 50 +++++++++++++++++-----
.../segmented/SegmentedRaftLogOutputStream.java | 12 +++---
.../raftlog/segmented/SegmentedRaftLogReader.java | 28 +++++++-----
.../raftlog/segmented/SegmentedRaftLogWorker.java | 50 +++++++++++-----------
.../org/apache/ratis/server/storage/MetaFile.java | 6 +--
.../apache/ratis/server/storage/RaftStorage.java | 6 +--
.../ratis/server/storage/RaftStorageDirectory.java | 15 +++++--
.../apache/ratis/statemachine/StateMachine.java | 3 +-
.../ratis/statemachine/TransactionContext.java | 3 +-
.../ratis/statemachine/impl/BaseStateMachine.java | 24 ++++++++---
.../impl/SimpleStateMachineStorage.java | 4 +-
.../statemachine/impl/TransactionContextImpl.java | 12 +++---
.../statemachine/SimpleStateMachine4Testing.java | 8 ++--
.../raftlog/segmented/TestSegmentedRaftLog.java | 8 ++--
.../segmented/TestSegmentedRaftLogCache.java | 46 ++++++++++----------
.../java/org/apache/ratis/tools/ParseRatisLog.java | 2 +-
35 files changed, 293 insertions(+), 200 deletions(-)
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
index 94f6b1c..16aeb40 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
@@ -94,9 +94,9 @@ public class GrpcLogAppender extends LogAppender {
final long nextIndex = 1 + Optional.ofNullable(request)
.map(AppendEntriesRequest::getPreviousLog)
.map(TermIndex::getIndex)
- .orElseGet(follower::getMatchIndex);
+ .orElseGet(getFollower()::getMatchIndex);
pendingRequests.clear();
- follower.decreaseNextIndex(nextIndex);
+ getFollower().decreaseNextIndex(nextIndex);
}
@Override
@@ -133,7 +133,7 @@ public class GrpcLogAppender extends LogAppender {
if (!shouldSendRequest()) {
return getHeartbeatRemainingTime(); // No requests, wait until heartbeat
} else if (shouldWait()) {
- return halfMinTimeoutMs; // Should wait for a short time
+ return getHalfMinTimeoutMs(); // Should wait for a short time
}
return 0L;
}
@@ -198,13 +198,13 @@ public class GrpcLogAppender extends LogAppender {
private void sendRequest(AppendEntriesRequest request, AppendEntriesRequestProto proto,
StreamObserver<AppendEntriesRequestProto> s) {
CodeInjectionForTesting.execute(GrpcService.GRPC_SEND_SERVER_REQUEST,
- server.getId(), null, proto);
+ getServer().getId(), null, proto);
request.startRequestTimer();
s.onNext(proto);
scheduler.onTimeout(requestTimeoutDuration,
() -> timeoutAppendRequest(request.getCallId(), request.isHeartbeat()),
LOG, () -> "Timeout check failed for append entry request: " + request);
- follower.updateLastRpcSendTime();
+ getFollower().updateLastRpcSendTime();
}
private void timeoutAppendRequest(long cid, boolean heartbeat) {
@@ -218,7 +218,7 @@ public class GrpcLogAppender extends LogAppender {
private void increaseNextIndex(AppendEntriesRequestProto request) {
final int count = request.getEntriesCount();
if (count > 0) {
- follower.increaseNextIndex(request.getEntries(count - 1).getIndex() + 1);
+ getFollower().increaseNextIndex(request.getEntries(count - 1).getIndex() + 1);
}
}
@@ -226,7 +226,7 @@ public class GrpcLogAppender extends LogAppender {
* StreamObserver for handling responses from the follower
*/
private class AppendLogResponseHandler implements StreamObserver<AppendEntriesReplyProto> {
- private final String name = follower.getName() + "-" + getClass().getSimpleName();
+ private final String name = getFollower().getName() + "-" + getClass().getSimpleName();
/**
* After receiving a appendEntries reply, do the following:
@@ -259,7 +259,7 @@ public class GrpcLogAppender extends LogAppender {
private void onNextImpl(AppendEntriesReplyProto reply) {
// update the last rpc time
- follower.updateLastRpcResponseTime();
+ getFollower().updateLastRpcResponseTime();
if (!firstResponseReceived) {
firstResponseReceived = true;
@@ -269,7 +269,7 @@ public class GrpcLogAppender extends LogAppender {
case SUCCESS:
grpcServerMetrics.onRequestSuccess(getFollowerId().toString(), reply.getIsHearbeat());
updateCommitIndex(reply.getFollowerCommit());
- if (follower.updateMatchIndex(reply.getMatchIndex())) {
+ if (getFollower().updateMatchIndex(reply.getMatchIndex())) {
submitEventOnSuccessAppend();
}
break;
@@ -318,11 +318,11 @@ public class GrpcLogAppender extends LogAppender {
private synchronized void updateNextIndex(long replyNextIndex) {
pendingRequests.clear();
- follower.updateNextIndex(replyNextIndex);
+ getFollower().updateNextIndex(replyNextIndex);
}
private class InstallSnapshotResponseHandler implements StreamObserver<InstallSnapshotReplyProto> {
- private final String name = follower.getName() + "-" + getClass().getSimpleName();
+ private final String name = getFollower().getName() + "-" + getClass().getSimpleName();
private final Queue<Integer> pending;
private final AtomicBoolean done = new AtomicBoolean(false);
@@ -361,7 +361,7 @@ public class GrpcLogAppender extends LogAppender {
}
// update the last rpc time
- follower.updateLastRpcResponseTime();
+ getFollower().updateLastRpcResponseTime();
if (!firstResponseReceived) {
firstResponseReceived = true;
@@ -375,7 +375,7 @@ public class GrpcLogAppender extends LogAppender {
case ALREADY_INSTALLED:
final long followerSnapshotIndex = reply.getSnapshotIndex();
LOG.info("{}: set follower snapshotIndex to {}.", this, followerSnapshotIndex);
- follower.setSnapshotIndex(followerSnapshotIndex);
+ getFollower().setSnapshotIndex(followerSnapshotIndex);
removePending(reply);
break;
case NOT_LEADER:
@@ -384,7 +384,7 @@ public class GrpcLogAppender extends LogAppender {
case CONF_MISMATCH:
LOG.error("{}: Configuration Mismatch ({}): Leader {} has it set to {} but follower {} has it set to {}",
this, RaftServerConfigKeys.Log.Appender.INSTALL_SNAPSHOT_ENABLED_KEY,
- server.getId(), installSnapshotEnabled, getFollowerId(), !installSnapshotEnabled);
+ getServer().getId(), installSnapshotEnabled, getFollowerId(), !installSnapshotEnabled);
case UNRECOGNIZED:
default:
break;
@@ -420,7 +420,7 @@ public class GrpcLogAppender extends LogAppender {
*/
private void installSnapshot(SnapshotInfo snapshot) {
LOG.info("{}: followerNextIndex = {} but logStartIndex = {}, send snapshot {} to follower",
- this, follower.getNextIndex(), raftLog.getStartIndex(), snapshot);
+ this, getFollower().getNextIndex(), getRaftLog().getStartIndex(), snapshot);
final InstallSnapshotResponseHandler responseHandler = new InstallSnapshotResponseHandler();
StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver = null;
@@ -431,7 +431,7 @@ public class GrpcLogAppender extends LogAppender {
new SnapshotRequestIter(snapshot, requestId)) {
if (isAppenderRunning()) {
snapshotRequestObserver.onNext(request);
- follower.updateLastRpcSendTime();
+ getFollower().updateLastRpcSendTime();
responseHandler.addPending(request);
} else {
break;
@@ -457,7 +457,7 @@ public class GrpcLogAppender extends LogAppender {
}
if (responseHandler.hasAllResponse()) {
- follower.setSnapshotIndex(snapshot.getTermIndex().getIndex());
+ getFollower().setSnapshotIndex(snapshot.getTermIndex().getIndex());
LOG.info("{}: installed snapshot {} successfully", this, snapshot);
}
}
@@ -468,7 +468,7 @@ public class GrpcLogAppender extends LogAppender {
*/
private void installSnapshot(TermIndex firstAvailableLogTermIndex) {
LOG.info("{}: followerNextIndex = {} but logStartIndex = {}, notify follower to install snapshot-{}",
- this, follower.getNextIndex(), raftLog.getStartIndex(), firstAvailableLogTermIndex);
+ this, getFollower().getNextIndex(), getRaftLog().getStartIndex(), firstAvailableLogTermIndex);
final InstallSnapshotResponseHandler responseHandler = new InstallSnapshotResponseHandler();
StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver = null;
@@ -480,7 +480,7 @@ public class GrpcLogAppender extends LogAppender {
try {
snapshotRequestObserver = getClient().installSnapshot(responseHandler);
snapshotRequestObserver.onNext(request);
- follower.updateLastRpcSendTime();
+ getFollower().updateLastRpcSendTime();
responseHandler.addPending(request);
snapshotRequestObserver.onCompleted();
} catch (Exception e) {
@@ -507,12 +507,12 @@ public class GrpcLogAppender extends LogAppender {
* @return the first available log's start term index
*/
private TermIndex shouldNotifyToInstallSnapshot() {
- if (follower.getNextIndex() < raftLog.getStartIndex()) {
+ if (getFollower().getNextIndex() < getRaftLog().getStartIndex()) {
// The Leader does not have the logs from the Follower's last log
// index onwards. And install snapshot is disabled. So the Follower
// should be notified to install the latest snapshot through its
// State Machine.
- return raftLog.getTermIndex(raftLog.getStartIndex());
+ return getRaftLog().getTermIndex(getRaftLog().getStartIndex());
}
return null;
}
diff --git a/ratis-logservice/src/main/java/org/apache/ratis/logservice/server/LogStateMachine.java b/ratis-logservice/src/main/java/org/apache/ratis/logservice/server/LogStateMachine.java
index ba86d01..1d13457 100644
--- a/ratis-logservice/src/main/java/org/apache/ratis/logservice/server/LogStateMachine.java
+++ b/ratis-logservice/src/main/java/org/apache/ratis/logservice/server/LogStateMachine.java
@@ -182,7 +182,7 @@ public class LogStateMachine extends BaseStateMachine {
private void checkInitialization() throws IOException {
if (this.log == null) {
- ServerState serverState = proxy.getImpl(groupId).getState();
+ ServerState serverState = proxy.getImpl(getGroupId()).getState();
this.log = serverState.getLog();
}
}
@@ -752,8 +752,8 @@ public class LogStateMachine extends BaseStateMachine {
private RaftClient getClient() throws IOException {
if (client == null) {
try {
- RaftServer raftServer = server.get();
- client = RaftClient.newBuilder().setRaftGroup(getGroupFromGroupId(raftServer, groupId))
+ RaftServer raftServer = getServer().get();
+ client = RaftClient.newBuilder().setRaftGroup(getGroupFromGroupId(raftServer, getGroupId()))
.setClientId(ClientId.randomId())
.setProperties(raftServer.getProperties()).build();
} catch (Exception e) {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/RaftServer.java b/ratis-server/src/main/java/org/apache/ratis/server/RaftServer.java
index d7deb15..c50a80a 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/RaftServer.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/RaftServer.java
@@ -76,7 +76,8 @@ public interface RaftServer extends Closeable, RpcType.Get,
return ServerImplUtils.newRaftServer(
serverId,
group,
- Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' is initialized."),
+ Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
+ "is initialized."),
Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
parameters);
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/RaftServerRpc.java b/ratis-server/src/main/java/org/apache/ratis/server/RaftServerRpc.java
index e75c340..ee34d88 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/RaftServerRpc.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/RaftServerRpc.java
@@ -42,8 +42,8 @@ public interface RaftServerRpc extends RaftServerProtocol, RpcType.Get, Closeabl
"The 'server' field is not initialized.");
}
- public B setServer(RaftServer server) {
- this.server = server;
+ public B setServer(RaftServer raftServer) {
+ this.server = raftServer;
return getThis();
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
index 1148db8..4c05a09 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
@@ -72,8 +72,8 @@ class LeaderElection implements Runnable {
enum Result {PASSED, REJECTED, TIMEOUT, DISCOVERED_A_NEW_TERM, SHUTDOWN}
private static class ResultAndTerm {
- final Result result;
- final long term;
+ private final Result result;
+ private final long term;
ResultAndTerm(Result result, long term) {
this.result = result;
@@ -246,8 +246,9 @@ class LeaderElection implements Runnable {
final long term = Math.max(r.term, state.getCurrentTerm());
server.changeToFollowerAndPersistMetadata(term, Result.DISCOVERED_A_NEW_TERM);
return;
- case TIMEOUT:
- // should start another election
+ case TIMEOUT: // should start another election
+ return;
+ default: throw new IllegalArgumentException("Unable to process result " + r.result);
}
}
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
index 9e53bfd..962c202 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
@@ -67,9 +67,9 @@ public class LeaderState {
STEP_DOWN, UPDATE_COMMIT, CHECK_STAGING
}
- final Type type;
- final long newTerm;
- final Runnable handler;
+ private final Type type;
+ private final long newTerm;
+ private final Runnable handler;
StateUpdateEvent(Type type, long newTerm, Runnable handler) {
this.type = type;
@@ -93,6 +93,11 @@ public class LeaderState {
}
@Override
+ public int hashCode() {
+ return Objects.hash(type, newTerm);
+ }
+
+ @Override
public String toString() {
return type + (newTerm >= 0? ":" + newTerm: "");
}
@@ -126,7 +131,9 @@ public class LeaderState {
if (e != null) {
// remove duplicated events from the head.
- for(; e.equals(queue.peek()); queue.poll());
+ while(e.equals(queue.peek())) {
+ queue.poll();
+ }
}
return e;
}
@@ -175,9 +182,9 @@ public class LeaderState {
}
}
- private final StateUpdateEvent UPDATE_COMMIT_EVENT =
+ private final StateUpdateEvent updateCommitEvent =
new StateUpdateEvent(StateUpdateEvent.Type.UPDATE_COMMIT, -1, this::updateCommit);
- private final StateUpdateEvent CHECK_STAGING_EVENT =
+ private final StateUpdateEvent checkStagingEvent =
new StateUpdateEvent(StateUpdateEvent.Type.CHECK_STAGING, -1, this::checkStaging);
private final String name;
@@ -298,11 +305,11 @@ public class LeaderState {
// add the request to the pending queue
final PendingRequest pending = pendingRequests.addConfRequest(request);
- ConfigurationStagingState stagingState = new ConfigurationStagingState(
+ ConfigurationStagingState configurationStagingState = new ConfigurationStagingState(
peersToBootStrap, new PeerConfiguration(peersInNewConf));
- Collection<RaftPeer> newPeers = stagingState.getNewPeers();
+ Collection<RaftPeer> newPeers = configurationStagingState.getNewPeers();
// set the staging state
- this.stagingState = stagingState;
+ this.stagingState = configurationStagingState;
if (newPeers.isEmpty()) {
applyOldNewConf();
@@ -535,13 +542,13 @@ public class LeaderState {
}
void submitCheckStagingEvent() {
- eventQueue.submit(CHECK_STAGING_EVENT);
+ eventQueue.submit(checkStagingEvent);
}
private void checkStaging() {
if (!inStagingState()) {
// it is possible that the bootstrapping is done. Then, fallback to UPDATE_COMMIT
- UPDATE_COMMIT_EVENT.execute();
+ updateCommitEvent.execute();
} else {
final long committedIndex = server.getState().getLog()
.getLastCommittedIndex();
@@ -561,7 +568,7 @@ public class LeaderState {
}
void submitUpdateCommitEvent() {
- eventQueue.submit(UPDATE_COMMIT_EVENT);
+ eventQueue.submit(updateCommitEvent);
}
static class MinMajorityMax {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java
index 32339f2..c73693c 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java
@@ -120,14 +120,14 @@ public class LogAppender {
}
private final String name;
- protected final RaftServerImpl server;
+ private final RaftServerImpl server;
private final LeaderState leaderState;
- protected final RaftLog raftLog;
- protected final FollowerInfo follower;
+ private final RaftLog raftLog;
+ private final FollowerInfo follower;
private final DataQueue<EntryWithData> buffer;
private final int snapshotChunkMaxSize;
- protected final long halfMinTimeoutMs;
+ private final long halfMinTimeoutMs;
private final AppenderDaemon daemon;
@@ -148,6 +148,18 @@ public class LogAppender {
this.daemon = new AppenderDaemon();
}
+ public RaftServerImpl getServer() {
+ return server;
+ }
+
+ public RaftLog getRaftLog() {
+ return raftLog;
+ }
+
+ public long getHalfMinTimeoutMs() {
+ return halfMinTimeoutMs;
+ }
+
@Override
public String toString() {
return name;
@@ -480,7 +492,7 @@ public class LogAppender {
}
}
- private void handleReply(AppendEntriesReplyProto reply) {
+ private void handleReply(AppendEntriesReplyProto reply) throws IllegalArgumentException {
if (reply != null) {
switch (reply.getResult()) {
case SUCCESS:
@@ -508,6 +520,7 @@ public class LogAppender {
case UNRECOGNIZED:
LOG.warn("{}: received {}", this, reply.getResult());
break;
+ default: throw new IllegalArgumentException("Unable to process result " + reply.getResult());
}
}
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
index 42dcf8f..4cd0a0e 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/PendingRequests.java
@@ -144,7 +144,8 @@ class PendingRequests {
return r;
}
- Collection<TransactionContext> setNotLeaderException(NotLeaderException nle, Collection<CommitInfoProto> commitInfos) {
+ Collection<TransactionContext> setNotLeaderException(NotLeaderException nle,
+ Collection<CommitInfoProto> commitInfos) {
synchronized (this) {
resource.close();
permits.clear();
@@ -236,7 +237,8 @@ class PendingRequests {
* The leader state is stopped. Send NotLeaderException to all the pending
* requests since they have not got applied to the state machine yet.
*/
- Collection<TransactionContext> sendNotLeaderResponses(NotLeaderException nle, Collection<CommitInfoProto> commitInfos) {
+ Collection<TransactionContext> sendNotLeaderResponses(NotLeaderException nle,
+ Collection<CommitInfoProto> commitInfos) {
LOG.info("{}: sendNotLeaderResponses", name);
final Collection<TransactionContext> transactions = pendingRequests.setNotLeaderException(nle, commitInfos);
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftConfiguration.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftConfiguration.java
index 13c9afd..a56bd5a 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftConfiguration.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftConfiguration.java
@@ -33,14 +33,14 @@ import java.util.stream.Collectors;
*
* The objects of this class are immutable.
*/
-public class RaftConfiguration {
+public final class RaftConfiguration {
/** Create a {@link Builder}. */
public static Builder newBuilder() {
return new Builder();
}
/** To build {@link RaftConfiguration} objects. */
- public static class Builder {
+ public static final class Builder {
private PeerConfiguration oldConf;
private PeerConfiguration conf;
private long logEntryIndex = RaftServerConstants.INVALID_LOG_INDEX;
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerConstants.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerConstants.java
index da05037..0061ef4 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerConstants.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerConstants.java
@@ -19,13 +19,17 @@ package org.apache.ratis.server.impl;
import org.apache.ratis.server.raftlog.RaftLog;
-public interface RaftServerConstants {
+public final class RaftServerConstants {
/** @deprecated use {@link RaftLog#INVALID_LOG_INDEX}. */
@Deprecated
- long INVALID_LOG_INDEX = RaftLog.INVALID_LOG_INDEX;
- long DEFAULT_CALLID = 0;
+ public static final long INVALID_LOG_INDEX = RaftLog.INVALID_LOG_INDEX;
+ public static final long DEFAULT_CALLID = 0;
- enum StartupOption {
+ private RaftServerConstants() {
+ //Never constructed
+ }
+
+ public enum StartupOption {
FORMAT("format"),
REGULAR("regular");
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index f676c5c..e5b37b9 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -80,7 +80,8 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
private final LifeCycle lifeCycle;
private final ServerState state;
- private final Supplier<RaftPeer> peerSupplier = JavaUtils.memoize(() -> new RaftPeer(getId(), getServerRpc().getInetSocketAddress()));
+ private final Supplier<RaftPeer> peerSupplier = JavaUtils.memoize(() ->
+ new RaftPeer(getId(), getServerRpc().getInetSocketAddress()));
private final RoleInfo role;
private final RetryCache retryCache;
@@ -124,11 +125,11 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
}
LogAppender newLogAppender(
- LeaderState state, RaftPeer peer, Timestamp lastRpcTime, long nextIndex,
+ LeaderState leaderState, RaftPeer peer, Timestamp lastRpcTime, long nextIndex,
boolean attendVote) {
final FollowerInfo f = new FollowerInfo(getMemberId(), peer, lastRpcTime, nextIndex, attendVote,
rpcSlownessTimeoutMs);
- return getProxy().getFactory().newLogAppender(this, state, f);
+ return getProxy().getFactory().newLogAppender(this, leaderState, f);
}
RaftPeer getPeer() {
@@ -571,7 +572,6 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
return reply;
}
// let the state machine handle read-only request from client
- final StateMachine stateMachine = getStateMachine();
RaftClientRequest.Type type = request.getType();
if (type.is(RaftClientRequestProto.TypeCase.STREAM)) {
if (type.getStream().getClose()) {
@@ -644,7 +644,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
return CompletableFuture.completedFuture(
new RaftClientReply(request, new StateMachineException(getMemberId(), e), getCommitInfos()));
}
- return processQueryFuture(getStateMachine().queryStale(request.getMessage(), minIndex), request);
+ return processQueryFuture(stateMachine.queryStale(request.getMessage(), minIndex), request);
}
private CompletableFuture<RaftClientReply> streamAsync(RaftClientRequest request) {
@@ -946,6 +946,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
}
}
+ @SuppressWarnings("checkstyle:parameternumber")
private CompletableFuture<AppendEntriesReplyProto> appendEntriesAsync(
RaftPeerId leaderId, long leaderTerm, TermIndex previous, long leaderCommit, long callId, boolean initializing,
List<CommitInfoProto> commitInfos, LogEntryProto... entries) {
@@ -1329,7 +1330,6 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
}
CompletableFuture<Message> applyLogToStateMachine(LogEntryProto next) {
- final StateMachine stateMachine = getStateMachine();
if (!next.hasStateMachineLogEntry()) {
stateMachine.notifyIndexUpdate(next.getTerm(), next.getIndex());
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java
index a4fc4ad..e2e27f5 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java
@@ -382,7 +382,7 @@ public class RaftServerProxy implements RaftServer {
Preconditions.assertTrue(started, () -> getId()+ ": failed to start a new impl: " + newImpl);
return new RaftClientReply(request, newImpl.getCommitInfos());
}, implExecutor)
- .whenComplete((_1, throwable) -> {
+ .whenComplete((raftClientReply, throwable) -> {
if (throwable != null) {
if (!(throwable.getCause() instanceof AlreadyExistsException)) {
impls.remove(newGroup.getGroupId());
@@ -400,7 +400,8 @@ public class RaftServerProxy implements RaftServer {
RaftClientRequest request, RaftGroupId groupId, boolean deleteDirectory) {
if (!request.getRaftGroupId().equals(groupId)) {
return JavaUtils.completeExceptionally(new GroupMismatchException(
- getId() + ": Request group id (" + request.getRaftGroupId() + ") does not match the given group id " + groupId));
+ getId() + ": Request group id (" + request.getRaftGroupId() + ") does not match the given group id " +
+ groupId));
}
final CompletableFuture<RaftServerImpl> f = impls.remove(groupId);
if (f == null) {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java
index 8f91c23..24e05cd 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java
@@ -31,7 +31,10 @@ import java.io.IOException;
import java.util.concurrent.TimeUnit;
/** Server utilities for internal use. */
-public class ServerImplUtils {
+public final class ServerImplUtils {
+ private ServerImplUtils() {
+ //Never constructed
+ }
/** For the case that all {@link RaftServerImpl} objects share the same {@link StateMachine}. */
public static RaftServerProxy newRaftServer(
RaftPeerId id, RaftGroup group, StateMachine.Registry stateMachineRegistry,
@@ -42,7 +45,7 @@ public class ServerImplUtils {
return proxy;
}
- private static RaftServerProxy newRaftServer(
+ public static RaftServerProxy newRaftServer(
RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
throws IOException {
final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerProtoUtils.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerProtoUtils.java
index 0bfb342..143655c 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerProtoUtils.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerProtoUtils.java
@@ -371,6 +371,7 @@ public interface ServerProtoUtils {
return builder.build();
}
+ @SuppressWarnings("checkstyle:parameternumber")
static InstallSnapshotRequestProto toInstallSnapshotRequestProto(
RaftGroupMemberId requestorId, RaftPeerId replyId, String requestId, int requestIndex,
long term, TermIndex lastTermIndex, List<FileChunkProto> chunks,
@@ -423,6 +424,7 @@ public interface ServerProtoUtils {
.build();
}
+ @SuppressWarnings("checkstyle:parameternumber")
static AppendEntriesRequestProto toAppendEntriesRequestProto(
RaftGroupMemberId requestorId, RaftPeerId replyId, long leaderTerm,
List<LogEntryProto> entries, long leaderCommit, boolean initializing,
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java
index d4e8924..3ed045a 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java
@@ -288,7 +288,7 @@ public class ServerState implements Closeable {
* If accept, update the current state.
* @return true if the check passes
*/
- boolean recognizeLeader(RaftPeerId leaderId, long leaderTerm) {
+ boolean recognizeLeader(RaftPeerId peerLeaderId, long leaderTerm) {
final long current = currentTerm.get();
if (leaderTerm < current) {
return false;
@@ -298,7 +298,7 @@ public class ServerState implements Closeable {
// leader and term later
return true;
}
- return this.leaderId.equals(leaderId);
+ return this.leaderId.equals(peerLeaderId);
}
/**
@@ -364,16 +364,16 @@ public class ServerState implements Closeable {
}
}
- boolean updateStatemachine(long majorityIndex, long currentTerm) {
- if (log.updateLastCommitted(majorityIndex, currentTerm)) {
+ boolean updateStatemachine(long majorityIndex, long curTerm) {
+ if (log.updateLastCommitted(majorityIndex, curTerm)) {
stateMachineUpdater.notifyUpdater();
return true;
}
return false;
}
- void reloadStateMachine(long lastIndexInSnapshot, long currentTerm) {
- log.updateLastCommitted(lastIndexInSnapshot, currentTerm);
+ void reloadStateMachine(long lastIndexInSnapshot, long curTerm) {
+ log.updateLastCommitted(lastIndexInSnapshot, curTerm);
stateMachineUpdater.reloadStateMachine();
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java
index c578842..50b56e8 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java
@@ -261,10 +261,10 @@ class StateMachineUpdater implements Runnable {
i = stateMachine.takeSnapshot();
takeSnapshotTimerContext.stop();
- final long appliedIndex = getLastAppliedIndex();
- if (i > appliedIndex) {
+ final long lastAppliedIndex = getLastAppliedIndex();
+ if (i > lastAppliedIndex) {
throw new StateMachineException(
- "Bug in StateMachine: snapshot index = " + i + " > appliedIndex = " + appliedIndex
+ "Bug in StateMachine: snapshot index = " + i + " > appliedIndex = " + lastAppliedIndex
+ "; StateMachine class=" + stateMachine.getClass().getName() + ", stateMachine=" + stateMachine);
}
stateMachine.getStateMachineStorage().cleanupOldSnapshots(snapshotRetentionPolicy);
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLog.java
index a9adb96..6739478 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLog.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLog.java
@@ -316,7 +316,8 @@ public abstract class RaftLog implements RaftLogSequentialOps, Closeable {
Preconditions.assertTrue(entry.getTerm() >= lastTermIndex.getTerm(),
"Entry term less than RaftLog's last term: %d, entry: %s", lastTermIndex.getTerm(), entry);
Preconditions.assertTrue(entry.getIndex() == lastTermIndex.getIndex() + 1,
- "Difference between entry index and RaftLog's last index %d greater than 1, entry: %s", lastTermIndex.getIndex(), entry);
+ "Difference between entry index and RaftLog's last index %d greater than 1, entry: %s",
+ lastTermIndex.getIndex(), entry);
}
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java
index 1799fae..0d50341 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/memory/MemoryRaftLog.java
@@ -158,9 +158,9 @@ public class MemoryRaftLog extends RaftLog {
}
@Override
- public List<CompletableFuture<Long>> appendImpl(LogEntryProto... entries) {
+ public List<CompletableFuture<Long>> appendImpl(LogEntryProto... logEntryProtos) {
checkLogState();
- if (entries == null || entries.length == 0) {
+ if (logEntryProtos == null || logEntryProtos.length == 0) {
return Collections.emptyList();
}
try(AutoCloseableLock writeLock = writeLock()) {
@@ -173,26 +173,26 @@ public class MemoryRaftLog extends RaftLog {
// the quorum of peers' disks.
// TODO add a unit test for this
boolean toTruncate = false;
- int truncateIndex = (int) entries[0].getIndex();
+ int truncateIndex = (int) logEntryProtos[0].getIndex();
int index = 0;
- for (; truncateIndex < getNextIndex() && index < entries.length;
+ for (; truncateIndex < getNextIndex() && index < logEntryProtos.length;
index++, truncateIndex++) {
if (this.entries.get(truncateIndex).getTerm() !=
- entries[index].getTerm()) {
+ logEntryProtos[index].getTerm()) {
toTruncate = true;
break;
}
}
final List<CompletableFuture<Long>> futures;
if (toTruncate) {
- futures = new ArrayList<>(entries.length - index + 1);
+ futures = new ArrayList<>(logEntryProtos.length - index + 1);
futures.add(truncate(truncateIndex));
} else {
- futures = new ArrayList<>(entries.length - index);
+ futures = new ArrayList<>(logEntryProtos.length - index);
}
- for (int i = index; i < entries.length; i++) {
- this.entries.add(entries[i]);
- futures.add(CompletableFuture.completedFuture(entries[i].getIndex()));
+ for (int i = index; i < logEntryProtos.length; i++) {
+ this.entries.add(logEntryProtos[i]);
+ futures.add(CompletableFuture.completedFuture(logEntryProtos[i].getIndex()));
}
return futures;
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java
index 20a9afd..ad50ba9 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java
@@ -51,6 +51,11 @@ import java.util.function.Consumer;
* This class will be protected by the {@link SegmentedRaftLog}'s read-write lock.
*/
public class LogSegment implements Comparable<Long> {
+
+ //TODO: This class needs to be made final to address checkstyle issue. However, TestCacheEviction fails as Mockito
+ // cannot spy final class. This problem can be fixed when stable version of Mockito 2.x is available which provides
+ // a feature to mock final class/method.
+
static final Logger LOG = LoggerFactory.getLogger(LogSegment.class);
static long getEntrySize(LogEntryProto entry) {
@@ -215,10 +220,10 @@ public class LogSegment implements Comparable<Long> {
private volatile boolean isOpen;
private long totalSize = SegmentedRaftLogFormat.getHeaderLength();
/** Segment start index, inclusive. */
- private final long startIndex;
+ private long startIndex;
/** Segment end index, inclusive. */
private volatile long endIndex;
- private final RaftStorage storage;
+ private RaftStorage storage;
private RaftLogMetrics raftLogMetrics;
private final LogEntryLoader cacheLoader = new LogEntryLoader(raftLogMetrics);
/** later replace it with a metric */
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogCache.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogCache.java
index d43a315..70b4dc8 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogCache.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogCache.java
@@ -55,11 +55,31 @@ class SegmentedRaftLogCache {
return new SegmentFileInfo(ls.getStartIndex(), ls.getEndIndex(), ls.isOpen(), 0, 0);
}
- final long startIndex; // start index of the segment
- final long endIndex; // original end index
- final boolean isOpen;
- final long targetLength; // position for truncation
- final long newEndIndex; // new end index after the truncation
+ private final long startIndex; // start index of the segment
+ private final long endIndex; // original end index
+ private final boolean isOpen;
+ private final long targetLength; // position for truncation
+ private final long newEndIndex; // new end index after the truncation
+
+ public long getStartIndex() {
+ return startIndex;
+ }
+
+ public long getEndIndex() {
+ return endIndex;
+ }
+
+ public boolean isOpen() {
+ return isOpen;
+ }
+
+ public long getTargetLength() {
+ return targetLength;
+ }
+
+ public long getNewEndIndex() {
+ return newEndIndex;
+ }
private SegmentFileInfo(long start, long end, boolean isOpen, long targetLength, long newEndIndex) {
this.startIndex = start;
@@ -78,11 +98,19 @@ class SegmentedRaftLogCache {
}
static class TruncationSegments {
- final SegmentFileInfo toTruncate; // name of the file to be truncated
- final SegmentFileInfo[] toDelete; // names of the files to be deleted
+ private final SegmentFileInfo toTruncate; // name of the file to be truncated
+ private final SegmentFileInfo[] toDelete; // names of the files to be deleted
+
+ public SegmentFileInfo getToTruncate() {
+ return toTruncate;
+ }
+
+ public SegmentFileInfo[] getToDelete() {
+ return toDelete;
+ }
TruncationSegments(SegmentFileInfo toTruncate,
- List<SegmentFileInfo> toDelete) {
+ List<SegmentFileInfo> toDelete) {
this.toDelete = toDelete == null ? null :
toDelete.toArray(new SegmentFileInfo[toDelete.size()]);
this.toTruncate = toTruncate;
@@ -316,7 +344,7 @@ class SegmentedRaftLogCache {
void loadSegment(LogPathAndIndex pi, boolean keepEntryInCache,
Consumer<LogEntryProto> logConsumer) throws IOException {
LogSegment logSegment = LogSegment.loadSegment(storage, pi.getPath().toFile(),
- pi.startIndex, pi.endIndex, pi.isOpen(), keepEntryInCache, logConsumer, raftLogMetrics);
+ pi.getStartIndex(), pi.getEndIndex(), pi.isOpen(), keepEntryInCache, logConsumer, raftLogMetrics);
if (logSegment != null) {
addSegment(logSegment);
}
@@ -487,8 +515,8 @@ class SegmentedRaftLogCache {
}
static class TruncateIndices {
- final int arrayIndex;
- final long truncateIndex;
+ private final int arrayIndex;
+ private final long truncateIndex;
TruncateIndices(int arrayIndex, long truncateIndex) {
this.arrayIndex = arrayIndex;
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogOutputStream.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogOutputStream.java
index 0e910d7..10c40a4 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogOutputStream.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogOutputStream.java
@@ -36,14 +36,14 @@ import java.util.zip.Checksum;
public class SegmentedRaftLogOutputStream implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(SegmentedRaftLogOutputStream.class);
- private static final ByteBuffer fill;
+ private static final ByteBuffer FILL;
private static final int BUFFER_SIZE = 1024 * 1024; // 1 MB
static {
- fill = ByteBuffer.allocateDirect(BUFFER_SIZE);
- for (int i = 0; i < fill.capacity(); i++) {
- fill.put(SegmentedRaftLogFormat.getTerminator());
+ FILL = ByteBuffer.allocateDirect(BUFFER_SIZE);
+ for (int i = 0; i < FILL.capacity(); i++) {
+ FILL.put(SegmentedRaftLogFormat.getTerminator());
}
- fill.flip();
+ FILL.flip();
}
private final File file;
@@ -129,7 +129,7 @@ public class SegmentedRaftLogOutputStream implements Closeable {
private long preallocate(FileChannel fc, long outstanding) throws IOException {
final long actual = actualPreallocateSize(outstanding, segmentMaxSize - fc.size(), preallocatedSize);
Preconditions.assertTrue(actual >= outstanding);
- final long allocated = IOUtils.preallocate(fc, actual, fill);
+ final long allocated = IOUtils.preallocate(fc, actual, FILL);
LOG.debug("Pre-allocated {} bytes for {}", allocated, this);
return allocated;
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java
index 5caec22..b529ad0 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java
@@ -65,7 +65,9 @@ class SegmentedRaftLogReader implements Closeable {
public int read() throws IOException {
checkLimit(1);
int ret = super.read();
- if (ret != -1) curPos++;
+ if (ret != -1) {
+ curPos++;
+ }
return ret;
}
@@ -73,7 +75,9 @@ class SegmentedRaftLogReader implements Closeable {
public int read(byte[] data) throws IOException {
checkLimit(data.length);
int ret = super.read(data);
- if (ret > 0) curPos += ret;
+ if (ret > 0) {
+ curPos += ret;
+ }
return ret;
}
@@ -81,7 +85,9 @@ class SegmentedRaftLogReader implements Closeable {
public int read(byte[] data, int offset, int length) throws IOException {
checkLimit(length);
int ret = super.read(data, offset, length);
- if (ret > 0) curPos += ret;
+ if (ret > 0) {
+ curPos += ret;
+ }
return ret;
}
@@ -126,7 +132,7 @@ class SegmentedRaftLogReader implements Closeable {
}
}
- private static final int maxOpSize = 32 * 1024 * 1024;
+ private static final int MAX_OP_SIZE = 32 * 1024 * 1024;
private final File file;
private final LimitedInputStream limiter;
@@ -267,8 +273,8 @@ class SegmentedRaftLogReader implements Closeable {
* @return The log entry, or null if we hit EOF.
*/
private LogEntryProto decodeEntry() throws IOException {
- limiter.setLimit(maxOpSize);
- in.mark(maxOpSize);
+ limiter.setLimit(MAX_OP_SIZE);
+ in.mark(MAX_OP_SIZE);
byte nextByte;
try {
@@ -288,9 +294,9 @@ class SegmentedRaftLogReader implements Closeable {
// Here, we verify that the Op size makes sense and that the
// data matches its checksum before attempting to construct an Op.
int entryLength = CodedInputStream.readRawVarint32(nextByte, in);
- if (entryLength > maxOpSize) {
+ if (entryLength > MAX_OP_SIZE) {
throw new IOException("Entry has size " + entryLength
- + ", but maxOpSize = " + maxOpSize);
+ + ", but MAX_OP_SIZE = " + MAX_OP_SIZE);
}
final int varintLength = CodedOutputStream.computeUInt32SizeNoTag(
@@ -298,7 +304,7 @@ class SegmentedRaftLogReader implements Closeable {
final int totalLength = varintLength + entryLength;
checkBufferSize(totalLength);
in.reset();
- in.mark(maxOpSize);
+ in.mark(MAX_OP_SIZE);
IOUtils.readFully(in, temp, 0, totalLength);
// verify checksum
@@ -318,11 +324,11 @@ class SegmentedRaftLogReader implements Closeable {
}
private void checkBufferSize(int entryLength) {
- Preconditions.assertTrue(entryLength <= maxOpSize);
+ Preconditions.assertTrue(entryLength <= MAX_OP_SIZE);
int length = temp.length;
if (length < entryLength) {
while (length < entryLength) {
- length = Math.min(length * 2, maxOpSize);
+ length = Math.min(length * 2, MAX_OP_SIZE);
}
temp = new byte[length];
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
index 274a327..d8dc3d0 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
@@ -423,11 +423,11 @@ class SegmentedRaftLogWorker implements Runnable {
@Override
void execute() throws IOException {
- if (segments.toDelete != null) {
+ if (segments.getToDelete() != null) {
Timer.Context purgeLogContext = raftLogMetrics.getRaftLogPurgeTimer().time();
- for (SegmentFileInfo fileInfo : segments.toDelete) {
+ for (SegmentFileInfo fileInfo : segments.getToDelete()) {
File delFile = storage.getStorageDir()
- .getClosedLogFile(fileInfo.startIndex, fileInfo.endIndex);
+ .getClosedLogFile(fileInfo.getStartIndex(), fileInfo.getEndIndex());
FileUtils.deleteFile(delFile);
}
purgeLogContext.stop();
@@ -592,7 +592,7 @@ class SegmentedRaftLogWorker implements Runnable {
private class TruncateLog extends Task {
private final TruncationSegments segments;
private final long truncateIndex;
- CompletableFuture<Void> stateMachineFuture = null;
+ private CompletableFuture<Void> stateMachineFuture = null;
TruncateLog(TruncationSegments ts, long index) {
this.segments = ts;
@@ -611,46 +611,46 @@ class SegmentedRaftLogWorker implements Runnable {
void execute() throws IOException {
freeSegmentedRaftLogOutputStream();
- if (segments.toTruncate != null) {
- File fileToTruncate = segments.toTruncate.isOpen ?
+ if (segments.getToTruncate() != null) {
+ File fileToTruncate = segments.getToTruncate().isOpen() ?
storage.getStorageDir().getOpenLogFile(
- segments.toTruncate.startIndex) :
+ segments.getToTruncate().getStartIndex()) :
storage.getStorageDir().getClosedLogFile(
- segments.toTruncate.startIndex,
- segments.toTruncate.endIndex);
+ segments.getToTruncate().getStartIndex(),
+ segments.getToTruncate().getEndIndex());
Preconditions.assertTrue(fileToTruncate.exists(),
"File %s to be truncated does not exist", fileToTruncate);
- FileUtils.truncateFile(fileToTruncate, segments.toTruncate.targetLength);
+ FileUtils.truncateFile(fileToTruncate, segments.getToTruncate().getTargetLength());
// rename the file
File dstFile = storage.getStorageDir().getClosedLogFile(
- segments.toTruncate.startIndex, segments.toTruncate.newEndIndex);
+ segments.getToTruncate().getStartIndex(), segments.getToTruncate().getNewEndIndex());
Preconditions.assertTrue(!dstFile.exists(),
"Truncated file %s already exists ", dstFile);
FileUtils.move(fileToTruncate, dstFile);
LOG.info("{}: Truncated log file {} to length {} and moved it to {}", name,
- fileToTruncate, segments.toTruncate.targetLength, dstFile);
+ fileToTruncate, segments.getToTruncate().getTargetLength(), dstFile);
// update lastWrittenIndex
- lastWrittenIndex = segments.toTruncate.newEndIndex;
+ lastWrittenIndex = segments.getToTruncate().getNewEndIndex();
}
- if (segments.toDelete != null && segments.toDelete.length > 0) {
- long minStart = segments.toDelete[0].startIndex;
- for (SegmentFileInfo del : segments.toDelete) {
+ if (segments.getToDelete() != null && segments.getToDelete().length > 0) {
+ long minStart = segments.getToDelete()[0].getStartIndex();
+ for (SegmentFileInfo del : segments.getToDelete()) {
final File delFile;
- if (del.isOpen) {
- delFile = storage.getStorageDir().getOpenLogFile(del.startIndex);
+ if (del.isOpen()) {
+ delFile = storage.getStorageDir().getOpenLogFile(del.getStartIndex());
} else {
delFile = storage.getStorageDir()
- .getClosedLogFile(del.startIndex, del.endIndex);
+ .getClosedLogFile(del.getStartIndex(), del.getEndIndex());
}
Preconditions.assertTrue(delFile.exists(),
"File %s to be deleted does not exist", delFile);
FileUtils.deleteFile(delFile);
LOG.info("{}: Deleted log file {}", name, delFile);
- minStart = Math.min(minStart, del.startIndex);
+ minStart = Math.min(minStart, del.getStartIndex());
}
- if (segments.toTruncate == null) {
+ if (segments.getToTruncate() == null) {
lastWrittenIndex = minStart - 1;
}
}
@@ -663,10 +663,10 @@ class SegmentedRaftLogWorker implements Runnable {
@Override
long getEndIndex() {
- if (segments.toTruncate != null) {
- return segments.toTruncate.newEndIndex;
- } else if (segments.toDelete.length > 0) {
- return segments.toDelete[segments.toDelete.length - 1].endIndex;
+ if (segments.getToTruncate() != null) {
+ return segments.getToTruncate().getNewEndIndex();
+ } else if (segments.getToDelete().length > 0) {
+ return segments.getToDelete()[segments.getToDelete().length - 1].getEndIndex();
}
return RaftLog.INVALID_LOG_INDEX;
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/storage/MetaFile.java b/ratis-server/src/main/java/org/apache/ratis/server/storage/MetaFile.java
index 329f174..119ccbb 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/storage/MetaFile.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/storage/MetaFile.java
@@ -85,11 +85,11 @@ public class MetaFile {
*
* @throws IOException if the file cannot be written
*/
- void writeFile(long term, String votedFor) throws IOException {
+ void writeFile(long givenTerm, String votedForInfo) throws IOException {
AtomicFileOutputStream fos = new AtomicFileOutputStream(file);
Properties properties = new Properties();
- properties.setProperty(TERM_KEY, Long.toString(term));
- properties.setProperty(VOTEDFOR_KEY, votedFor);
+ properties.setProperty(TERM_KEY, Long.toString(givenTerm));
+ properties.setProperty(VOTEDFOR_KEY, votedForInfo);
try {
properties.store(
new BufferedWriter(new OutputStreamWriter(fos, StandardCharsets.UTF_8)), "");
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorage.java b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorage.java
index 0d407d7..f989875 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorage.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorage.java
@@ -86,9 +86,9 @@ public class RaftStorage implements Closeable {
}
private MetaFile writeMetaFile(long term, String votedFor) throws IOException {
- MetaFile metaFile = new MetaFile(storageDir.getMetaFile());
- metaFile.set(term, votedFor);
- return metaFile;
+ MetaFile mFile = new MetaFile(storageDir.getMetaFile());
+ mFile.set(term, votedFor);
+ return mFile;
}
private void cleanMetaTmpFile() throws IOException {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java
index 5d57a20..c9d8cc3 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java
@@ -63,8 +63,8 @@ public class RaftStorageDirectory {
public static class LogPathAndIndex {
private final Path path;
- public final long startIndex;
- public final long endIndex;
+ private final long startIndex;
+ private final long endIndex;
LogPathAndIndex(Path path, long startIndex, long endIndex) {
this.path = path;
@@ -72,6 +72,14 @@ public class RaftStorageDirectory {
this.endIndex = endIndex;
}
+ public long getStartIndex() {
+ return startIndex;
+ }
+
+ public long getEndIndex() {
+ return endIndex;
+ }
+
public Path getPath() {
return path;
}
@@ -359,8 +367,9 @@ public class RaftStorageDirectory {
* Unlock storage.
*/
public void unlock() throws IOException {
- if (this.lock == null)
+ if (this.lock == null) {
return;
+ }
this.lock.release();
lock.channel().close();
lock = null;
diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/StateMachine.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/StateMachine.java
index 4d2cd5d..db578ae 100644
--- a/ratis-server/src/main/java/org/apache/ratis/statemachine/StateMachine.java
+++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/StateMachine.java
@@ -24,7 +24,6 @@ import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftGroupMemberId;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.server.RaftServer;
-import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.impl.ServerProtoUtils;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.storage.RaftStorage;
@@ -88,7 +87,7 @@ public interface StateMachine extends Closeable {
*
* In the meanwhile, when the size of raft log outside of the latest snapshot
* exceeds certain threshold, the RaftServer may choose to trigger a snapshot
- * if {@link RaftServerConfigKeys.Snapshot#AUTO_TRIGGER_ENABLED_KEY} is enabled.
+ * if {@link org.apache.ratis.server.RaftServerConfigKeys.Snapshot#AUTO_TRIGGER_ENABLED_KEY} is enabled.
*
* The snapshot should include the latest raft configuration.
*
diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java
index fa1cd61..76e8509 100644
--- a/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java
+++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/TransactionContext.java
@@ -27,7 +27,6 @@ import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.apache.ratis.util.Preconditions;
import java.io.IOException;
-import java.util.Collection;
import java.util.Objects;
/**
@@ -42,7 +41,7 @@ import java.util.Objects;
* a {@link TransactionContext} with the changes from the {@link StateMachine}.
* The same context will be passed back to the {@link StateMachine}
* via the {@link StateMachine#applyTransaction(TransactionContext)} call
- * or the {@link StateMachine#notifyNotLeader(Collection)} call.
+ * or the {@link StateMachine#notifyNotLeader(java.util.Collection)} call.
*
* In the second case, the {@link StateMachine} is a follower.
* The {@link TransactionContext} will be a committed entry coming from
diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java
index 78583fa..bc4fc66 100644
--- a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java
+++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java
@@ -47,9 +47,9 @@ import java.util.concurrent.atomic.AtomicReference;
* Base implementation for StateMachines.
*/
public class BaseStateMachine implements StateMachine {
- protected final CompletableFuture<RaftServer> server = new CompletableFuture<>();
- protected volatile RaftGroupId groupId;
- protected final LifeCycle lifeCycle = new LifeCycle(getClass().getSimpleName());
+ private final CompletableFuture<RaftServer> server = new CompletableFuture<>();
+ private volatile RaftGroupId groupId;
+ private final LifeCycle lifeCycle = new LifeCycle(getClass().getSimpleName());
private final AtomicReference<TermIndex> lastAppliedTermIndex = new AtomicReference<>();
@@ -63,15 +63,27 @@ public class BaseStateMachine implements StateMachine {
return server.isDone()? server.join().getId(): null;
}
+ public LifeCycle getLifeCycle() {
+ return lifeCycle;
+ }
+
+ public CompletableFuture<RaftServer> getServer() {
+ return server;
+ }
+
+ public RaftGroupId getGroupId() {
+ return groupId;
+ }
+
@Override
public LifeCycle.State getLifeCycleState() {
return lifeCycle.getCurrentState();
}
@Override
- public void initialize(RaftServer server, RaftGroupId groupId, RaftStorage storage) throws IOException {
- this.groupId = groupId;
- this.server.complete(server);
+ public void initialize(RaftServer raftServer, RaftGroupId raftGroupId, RaftStorage storage) throws IOException {
+ this.groupId = raftGroupId;
+ this.server.complete(raftServer);
lifeCycle.setName("" + this);
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java
index db60f42..93e5470 100644
--- a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java
+++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/SimpleStateMachineStorage.java
@@ -60,8 +60,8 @@ public class SimpleStateMachineStorage implements StateMachineStorage {
private volatile SingleFileSnapshotInfo currentSnapshot = null;
@Override
- public void init(RaftStorage raftStorage) throws IOException {
- this.raftStorage = raftStorage;
+ public void init(RaftStorage rStorage) throws IOException {
+ this.raftStorage = rStorage;
this.smDir = raftStorage.getStorageDir().getStateMachineDir();
loadLatestSnapshot();
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java
index 59beecc..1cf9148 100644
--- a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/TransactionContextImpl.java
@@ -120,8 +120,8 @@ public class TransactionContextImpl implements TransactionContext {
}
@Override
- public TransactionContext setStateMachineContext(Object stateMachineContext) {
- this.stateMachineContext = stateMachineContext;
+ public TransactionContext setStateMachineContext(Object context) {
+ this.stateMachineContext = context;
return this;
}
@@ -139,8 +139,8 @@ public class TransactionContextImpl implements TransactionContext {
}
@Override
- public TransactionContext setStateMachineLogEntryProto(StateMachineLogEntryProto smLogEntryProto) {
- this.smLogEntryProto = smLogEntryProto;
+ public TransactionContext setStateMachineLogEntryProto(StateMachineLogEntryProto logEntryProto) {
+ this.smLogEntryProto = logEntryProto;
return this;
}
@@ -156,8 +156,8 @@ public class TransactionContextImpl implements TransactionContext {
}
@Override
- public TransactionContext setShouldCommit(boolean shouldCommit) {
- this.shouldCommit = shouldCommit;
+ public TransactionContext setShouldCommit(boolean sCommit) {
+ this.shouldCommit = sCommit;
return this;
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java b/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java
index 32ab38a..1e25728 100644
--- a/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java
+++ b/ratis-server/src/test/java/org/apache/ratis/statemachine/SimpleStateMachine4Testing.java
@@ -207,7 +207,7 @@ public class SimpleStateMachine4Testing extends BaseStateMachine {
RaftStorage raftStorage) throws IOException {
LOG.info("Initializing " + this);
this.groupId = groupId;
- lifeCycle.startAndTransition(() -> {
+ getLifeCycle().startAndTransition(() -> {
super.initialize(server, groupId, raftStorage);
storage.init(raftStorage);
loadSnapshot(storage.findLatestSnapshot());
@@ -222,8 +222,8 @@ public class SimpleStateMachine4Testing extends BaseStateMachine {
@Override
public synchronized void pause() {
- lifeCycle.transition(LifeCycle.State.PAUSING);
- lifeCycle.transition(LifeCycle.State.PAUSED);
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
}
@Override
@@ -379,7 +379,7 @@ public class SimpleStateMachine4Testing extends BaseStateMachine {
@Override
public void close() {
- lifeCycle.checkStateAndClose(() -> {
+ getLifeCycle().checkStateAndClose(() -> {
running = false;
checkpointer.interrupt();
});
diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
index 001fe73..f20dc5b 100644
--- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
+++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
@@ -549,8 +549,8 @@ public class TestSegmentedRaftLog extends BaseTest {
final StateMachine sm = new BaseStateMachine() {
@Override
public CompletableFuture<?> writeStateMachineData(LogEntryProto entry) {
- lifeCycle.transition(LifeCycle.State.STARTING);
- lifeCycle.transition(LifeCycle.State.RUNNING);
+ getLifeCycle().transition(LifeCycle.State.STARTING);
+ getLifeCycle().transition(LifeCycle.State.RUNNING);
return new CompletableFuture<>(); // the future never completes
}
@@ -562,8 +562,8 @@ public class TestSegmentedRaftLog extends BaseTest {
Assert.assertNotNull(entry);
- lifeCycle.transition(LifeCycle.State.PAUSING);
- lifeCycle.transition(LifeCycle.State.PAUSED);
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
}
};
diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLogCache.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLogCache.java
index 71c3dc6..a42b43d 100644
--- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLogCache.java
+++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLogCache.java
@@ -179,34 +179,34 @@ public class TestSegmentedRaftLogCache {
// check TruncationSegments
int currentNum= (int) (end / 100 + 1);
if (currentNum < numOfSegments) {
- Assert.assertEquals(1, ts.toDelete.length);
+ Assert.assertEquals(1, ts.getToDelete().length);
numOfSegments = currentNum;
} else {
- Assert.assertEquals(0, ts.toDelete.length);
+ Assert.assertEquals(0, ts.getToDelete().length);
}
}
// 230 entries remaining. truncate at the segment boundary
TruncationSegments ts = cache.truncate(200);
checkCache(0, 199, 100);
- Assert.assertEquals(1, ts.toDelete.length);
- Assert.assertEquals(200, ts.toDelete[0].startIndex);
- Assert.assertEquals(229, ts.toDelete[0].endIndex);
- Assert.assertEquals(0, ts.toDelete[0].targetLength);
- Assert.assertFalse(ts.toDelete[0].isOpen);
- Assert.assertNull(ts.toTruncate);
+ Assert.assertEquals(1, ts.getToDelete().length);
+ Assert.assertEquals(200, ts.getToDelete()[0].getStartIndex());
+ Assert.assertEquals(229, ts.getToDelete()[0].getEndIndex());
+ Assert.assertEquals(0, ts.getToDelete()[0].getTargetLength());
+ Assert.assertFalse(ts.getToDelete()[0].isOpen());
+ Assert.assertNull(ts.getToTruncate());
// add another open segment and truncate it as a whole
LogSegment newOpen = prepareLogSegment(200, 249, true);
cache.addSegment(newOpen);
ts = cache.truncate(200);
checkCache(0, 199, 100);
- Assert.assertEquals(1, ts.toDelete.length);
- Assert.assertEquals(200, ts.toDelete[0].startIndex);
- Assert.assertEquals(249, ts.toDelete[0].endIndex);
- Assert.assertEquals(0, ts.toDelete[0].targetLength);
- Assert.assertTrue(ts.toDelete[0].isOpen);
- Assert.assertNull(ts.toTruncate);
+ Assert.assertEquals(1, ts.getToDelete().length);
+ Assert.assertEquals(200, ts.getToDelete()[0].getStartIndex());
+ Assert.assertEquals(249, ts.getToDelete()[0].getEndIndex());
+ Assert.assertEquals(0, ts.getToDelete()[0].getTargetLength());
+ Assert.assertTrue(ts.getToDelete()[0].isOpen());
+ Assert.assertNull(ts.getToTruncate());
// add another open segment and truncate part of it
newOpen = prepareLogSegment(200, 249, true);
@@ -214,11 +214,11 @@ public class TestSegmentedRaftLogCache {
ts = cache.truncate(220);
checkCache(0, 219, 100);
Assert.assertNull(cache.getOpenSegment());
- Assert.assertEquals(0, ts.toDelete.length);
- Assert.assertTrue(ts.toTruncate.isOpen);
- Assert.assertEquals(219, ts.toTruncate.newEndIndex);
- Assert.assertEquals(200, ts.toTruncate.startIndex);
- Assert.assertEquals(249, ts.toTruncate.endIndex);
+ Assert.assertEquals(0, ts.getToDelete().length);
+ Assert.assertTrue(ts.getToTruncate().isOpen());
+ Assert.assertEquals(219, ts.getToTruncate().getNewEndIndex());
+ Assert.assertEquals(200, ts.getToTruncate().getStartIndex());
+ Assert.assertEquals(249, ts.getToTruncate().getEndIndex());
}
@Test
@@ -234,8 +234,8 @@ public class TestSegmentedRaftLogCache {
int purgeIndex = sIndex;
// open segment should never be purged
TruncationSegments ts = cache.purge(purgeIndex);
- Assert.assertNull(ts.toTruncate);
- Assert.assertEquals(end - start, ts.toDelete.length);
+ Assert.assertNull(ts.getToTruncate());
+ Assert.assertEquals(end - start, ts.getToDelete().length);
Assert.assertEquals(sIndex, cache.getStartIndex());
}
@@ -250,8 +250,8 @@ public class TestSegmentedRaftLogCache {
// overlapped close segment will not purged.
TruncationSegments ts = cache.purge(purgeIndex);
- Assert.assertNull(ts.toTruncate);
- Assert.assertEquals(end - start - 1, ts.toDelete.length);
+ Assert.assertNull(ts.getToTruncate());
+ Assert.assertEquals(end - start - 1, ts.getToDelete().length);
Assert.assertEquals(1, cache.getNumOfSegments());
}
diff --git a/ratis-tools/src/main/java/org/apache/ratis/tools/ParseRatisLog.java b/ratis-tools/src/main/java/org/apache/ratis/tools/ParseRatisLog.java
index 6ac3829..eddbaa7 100644
--- a/ratis-tools/src/main/java/org/apache/ratis/tools/ParseRatisLog.java
+++ b/ratis-tools/src/main/java/org/apache/ratis/tools/ParseRatisLog.java
@@ -56,7 +56,7 @@ public final class ParseRatisLog {
}
System.out.println("Processing Raft Log file: " + file.getAbsolutePath() + " size:" + file.length());
- final int entryCount = LogSegment.readSegmentFile(file, pi.startIndex, pi.endIndex, pi.isOpen(),
+ final int entryCount = LogSegment.readSegmentFile(file, pi.getStartIndex(), pi.getEndIndex(), pi.isOpen(),
RaftServerConfigKeys.Log.CorruptionPolicy.EXCEPTION, null, this::processLogEntry);
System.out.println("Num Total Entries: " + entryCount);
System.out.println("Num Conf Entries: " + numConfEntries);