You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by ru...@apache.org on 2020/11/05 05:54:12 UTC
[incubator-ratis] branch master updated: RATIS-950. Handle
Exceptions appropriately (#115)
This is an automated email from the ASF dual-hosted git repository.
runzhiwang 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 43a042a RATIS-950. Handle Exceptions appropriately (#115)
43a042a is described below
commit 43a042a8bbe123bcb5e567af0aeced12eb299290
Author: Dinesh Chitlangia <di...@gmail.com>
AuthorDate: Thu Nov 5 00:54:03 2020 -0500
RATIS-950. Handle Exceptions appropriately (#115)
---
.../java/org/apache/ratis/client/impl/OrderedAsync.java | 1 +
.../org/apache/ratis/client/impl/UnorderedAsync.java | 6 +++---
.../apache/ratis/retry/MultipleLinearRandomRetry.java | 2 +-
.../src/main/java/org/apache/ratis/rpc/RpcType.java | 8 ++++----
.../src/main/java/org/apache/ratis/util/FileUtils.java | 1 +
.../src/main/java/org/apache/ratis/util/IOUtils.java | 4 +++-
.../src/main/java/org/apache/ratis/util/LogUtils.java | 8 ++++----
.../java/org/apache/ratis/util/ResourceSemaphore.java | 4 ++--
.../src/main/java/org/apache/ratis/util/TaskQueue.java | 4 ++--
.../main/java/org/apache/ratis/util/TimeDuration.java | 3 ++-
.../ratis/grpc/client/GrpcClientProtocolClient.java | 6 +++---
.../ratis/grpc/client/GrpcClientProtocolService.java | 2 +-
.../org/apache/ratis/grpc/client/GrpcClientRpc.java | 9 +++++----
.../apache/ratis/grpc/client/GrpcClientStreamer.java | 5 +++++
.../org/apache/ratis/grpc/server/GrpcLogAppender.java | 7 +++++--
.../ratis/grpc/server/GrpcServerProtocolService.java | 6 +++---
.../java/org/apache/ratis/grpc/server/GrpcService.java | 3 ++-
.../apache/ratis/logservice/server/LogStateMachine.java | 1 +
.../main/java/org/apache/ratis/netty/NettyRpcProxy.java | 2 ++
.../org/apache/ratis/netty/server/NettyRpcService.java | 3 ++-
.../org/apache/ratis/server/impl/FollowerState.java | 5 +++--
.../org/apache/ratis/server/impl/LeaderElection.java | 2 +-
.../java/org/apache/ratis/server/impl/LeaderState.java | 5 ++++-
.../java/org/apache/ratis/server/impl/LogAppender.java | 5 ++++-
.../org/apache/ratis/server/impl/RaftServerImpl.java | 11 ++++++-----
.../org/apache/ratis/server/impl/RaftServerProxy.java | 4 ++--
.../org/apache/ratis/server/impl/ServerImplUtils.java | 1 +
.../java/org/apache/ratis/server/impl/ServerState.java | 1 +
.../java/org/apache/ratis/server/raftlog/RaftLog.java | 6 +++---
.../server/raftlog/segmented/SegmentedRaftLog.java | 8 ++++----
.../raftlog/segmented/SegmentedRaftLogInputStream.java | 6 +++---
.../raftlog/segmented/SegmentedRaftLogReader.java | 2 +-
.../raftlog/segmented/SegmentedRaftLogWorker.java | 17 +++++++++--------
.../src/test/java/org/apache/ratis/MiniRaftCluster.java | 7 ++++---
.../src/test/java/org/apache/ratis/RaftBasicTests.java | 2 +-
.../src/test/java/org/apache/ratis/RaftTestUtil.java | 2 +-
.../server/impl/BlockRequestHandlingInjection.java | 1 +
.../ratis/server/impl/DelayLocalExecutionInjection.java | 1 +
.../ratis/server/impl/RaftReconfigurationBaseTest.java | 1 +
.../ratis/server/impl/StateMachineShutdownTests.java | 1 +
.../apache/ratis/server/simulation/RequestHandler.java | 4 ++--
.../ratis/server/simulation/SimulatedRequestReply.java | 3 +++
.../ratis/server/simulation/SimulatedServerRpc.java | 1 +
.../ratis/statemachine/SimpleStateMachine4Testing.java | 8 ++++++--
.../apache/ratis/retry/TestExponentialBackoffRetry.java | 2 +-
.../org/apache/ratis/statemachine/TestStateMachine.java | 4 ++--
.../org/apache/ratis/util/TestDataBlockingQueue.java | 8 ++++----
.../org/apache/ratis/util/TestResourceSemaphore.java | 1 +
48 files changed, 124 insertions(+), 80 deletions(-)
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
index 0b642b7..bd14e23 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
@@ -158,6 +158,7 @@ public final class OrderedAsync {
try {
requestSemaphore.acquire();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
return JavaUtils.completeExceptionally(IOUtils.toInterruptedIOException(
"Interrupted when sending " + type + ", message=" + message, e));
}
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/UnorderedAsync.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/UnorderedAsync.java
index 5d02dbe..7b7f158 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/impl/UnorderedAsync.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/UnorderedAsync.java
@@ -122,9 +122,9 @@ public interface UnorderedAsync {
LOG.debug("schedule retry for attempt #{}, policy={}, request={}", attemptCount, retryPolicy, request);
client.getScheduler().onTimeout(sleepTime,
() -> sendRequestWithRetry(pending, client), LOG, () -> clientId + ": Failed~ to retry " + request);
- } catch (Throwable t) {
- LOG.error(clientId + ": Failed " + request, t);
- f.completeExceptionally(t);
+ } catch (Exception ex) {
+ LOG.error(clientId + ": Failed " + request, ex);
+ f.completeExceptionally(ex);
}
});
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/retry/MultipleLinearRandomRetry.java b/ratis-common/src/main/java/org/apache/ratis/retry/MultipleLinearRandomRetry.java
index 07816be..e273db5 100644
--- a/ratis-common/src/main/java/org/apache/ratis/retry/MultipleLinearRandomRetry.java
+++ b/ratis-common/src/main/java/org/apache/ratis/retry/MultipleLinearRandomRetry.java
@@ -180,7 +180,7 @@ public final class MultipleLinearRandomRetry implements RetryPolicy {
final String s = elements[i].trim().replace("_", "");
try {
return parser.apply(s);
- } catch(Throwable t) {
+ } catch(Exception t) {
LOG.warn("Failed to parse \"{}\", which is the index {} element in \"{}\"", s, i, input, t);
return null;
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/rpc/RpcType.java b/ratis-common/src/main/java/org/apache/ratis/rpc/RpcType.java
index 2d2f03a..1e412cd 100644
--- a/ratis-common/src/main/java/org/apache/ratis/rpc/RpcType.java
+++ b/ratis-common/src/main/java/org/apache/ratis/rpc/RpcType.java
@@ -33,20 +33,20 @@ public interface RpcType {
final Throwable fromSupportedRpcType;
try { // Try parsing it as a SupportedRpcType
return SupportedRpcType.valueOfIgnoreCase(rpcType);
- } catch (Throwable t) {
- fromSupportedRpcType = t;
+ } catch (Exception e) {
+ fromSupportedRpcType = e;
}
try {
// Try using it as a class name
return ReflectionUtils.newInstance(
ReflectionUtils.getClass(rpcType, RpcType.class));
- } catch(Throwable t) {
+ } catch(Exception e) {
final IllegalArgumentException iae = new IllegalArgumentException(
"Invalid " + RpcType.class.getSimpleName() + ": \"" + rpcType + "\" "
+ " cannot be used as a user-defined " + RpcType.class.getSimpleName()
+ " and it is not a " + SupportedRpcType.class.getSimpleName() + ".");
- iae.addSuppressed(t);
+ iae.addSuppressed(e);
iae.addSuppressed(fromSupportedRpcType);
throw iae;
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java b/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java
index eea2ea1..1b2492d 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java
@@ -39,6 +39,7 @@ public interface FileUtils {
try {
return JavaUtils.attempt(op, NUM_ATTEMPTS, SLEEP_TIME, name, LOG);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException("Interrupted " + name.get(), e);
}
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java b/ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java
index 16eb49b..899eb15 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java
@@ -65,6 +65,7 @@ public interface IOUtils {
try {
return future.get();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw toInterruptedIOException(name.get() + " interrupted.", e);
} catch (ExecutionException e) {
throw toIOException(e);
@@ -78,6 +79,7 @@ public interface IOUtils {
try {
return future.get(timeout.getDuration(), timeout.getUnit());
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw toInterruptedIOException(name.get() + " interrupted.", e);
} catch (ExecutionException e) {
throw toIOException(e);
@@ -194,7 +196,7 @@ public interface IOUtils {
if (c != null) {
try {
c.close();
- } catch(Throwable e) {
+ } catch(Exception e) {
if (log != null && log.isDebugEnabled()) {
log.debug("Exception in closing " + c, e);
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/LogUtils.java b/ratis-common/src/main/java/org/apache/ratis/util/LogUtils.java
index eb05ae1..9cb6e74 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/LogUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/LogUtils.java
@@ -56,13 +56,13 @@ public interface LogUtils {
final OUTPUT output;
try {
output = supplier.get();
- } catch (Throwable t) {
+ } catch (Exception e) {
if (log.isTraceEnabled()) {
- log.trace("Failed to " + name.get(), t);
+ log.trace("Failed to " + name.get(), e);
} else if (log.isWarnEnabled()){
- log.warn("Failed to " + name.get() + ": " + t);
+ log.warn("Failed to " + name.get() + ": " + e);
}
- final THROWABLE throwable = JavaUtils.cast(t);
+ final THROWABLE throwable = JavaUtils.cast(e);
throw throwable;
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java b/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
index 3380034..b9e0ff5 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/ResourceSemaphore.java
@@ -155,11 +155,11 @@ public class ResourceSemaphore extends Semaphore {
for (; i < permits.length; i++) {
resources.get(i).acquire(permits[i]);
}
- } catch (Throwable t) {
+ } catch (Exception e) {
for (; --i >= 0;) {
resources.get(i).release(permits[i]);
}
- throw t;
+ throw e;
}
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/TaskQueue.java b/ratis-common/src/main/java/org/apache/ratis/util/TaskQueue.java
index 47f902c..0089668 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/TaskQueue.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/TaskQueue.java
@@ -108,8 +108,8 @@ public class TaskQueue {
LOG.trace("{}: running {}", this, task);
try {
f.complete(task.get());
- } catch (Throwable e) {
- f.completeExceptionally(newThrowable.apply(e));
+ } catch (Throwable t) {
+ f.completeExceptionally(newThrowable.apply(t));
}
pollAndSubmit(executor);
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/TimeDuration.java b/ratis-common/src/main/java/org/apache/ratis/util/TimeDuration.java
index da78df7..951e5e9 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/TimeDuration.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/TimeDuration.java
@@ -313,7 +313,8 @@ public final class TimeDuration implements Comparable<TimeDuration> {
if (log != null) {
log.accept(StringUtils.stringSupplierAsObject(() -> "Completed sleeping " + this));
}
- } catch(InterruptedException ie) {
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
if (log != null) {
log.accept(StringUtils.stringSupplierAsObject(() -> "Interrupted sleeping " + this));
}
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
index 9e45db3..1c25ea6 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
@@ -284,8 +284,8 @@ public class GrpcClientProtocolClient implements Closeable {
return;
}
handleReplyFuture(callId, f -> f.complete(reply));
- } catch (Throwable t) {
- handleReplyFuture(callId, f -> f.completeExceptionally(t));
+ } catch (Exception e) {
+ handleReplyFuture(callId, f -> f.completeExceptionally(e));
}
}
@@ -316,7 +316,7 @@ public class GrpcClientProtocolClient implements Closeable {
if (!requestStreamer.onNext(ClientProtoUtils.toRaftClientRequestProto(request))) {
return JavaUtils.completeExceptionally(new AlreadyClosedException(getName() + ": the stream is closed."));
}
- } catch(Throwable t) {
+ } catch(Exception t) {
handleReplyFuture(request.getCallId(), future -> future.completeExceptionally(t));
return f;
}
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
index dc8ce49..7f31c9f 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
@@ -246,7 +246,7 @@ public class GrpcClientProtocolService extends RaftClientProtocolServiceImplBase
try {
final RaftClientRequest r = ClientProtoUtils.toRaftClientRequest(request);
processClientRequest(r);
- } catch (Throwable e) {
+ } catch (Exception e) {
responseError(e, () -> "onNext for " + ClientProtoUtils.toString(request) + " in " + name);
}
}
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
index 9fe24dd..a87b0a4 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
@@ -67,7 +67,7 @@ public class GrpcClientRpc extends RaftClientRpcWithProxy<GrpcClientProtocolClie
final GrpcClientProtocolClient proxy = getProxies().getProxy(serverId);
// Reuse the same grpc stream for all async calls.
return proxy.getOrderedStreamObservers().onNext(request);
- } catch (Throwable e) {
+ } catch (Exception e) {
return JavaUtils.completeExceptionally(e);
}
}
@@ -79,9 +79,9 @@ public class GrpcClientRpc extends RaftClientRpcWithProxy<GrpcClientProtocolClie
final GrpcClientProtocolClient proxy = getProxies().getProxy(serverId);
// Reuse the same grpc stream for all async calls.
return proxy.getUnorderedAsyncStreamObservers().onNext(request);
- } catch (Throwable t) {
- LOG.error(clientId + ": XXX Failed " + request, t);
- return JavaUtils.completeExceptionally(t);
+ } catch (Exception e) {
+ LOG.error(clientId + ": XXX Failed " + request, e);
+ return JavaUtils.completeExceptionally(e);
}
}
@@ -112,6 +112,7 @@ public class GrpcClientRpc extends RaftClientRpcWithProxy<GrpcClientProtocolClie
try {
return f.get();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new InterruptedIOException(
"Interrupted while waiting for response of request " + request);
} catch (ExecutionException e) {
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientStreamer.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientStreamer.java
index ccb182f..c6d86ef 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientStreamer.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientStreamer.java
@@ -154,6 +154,7 @@ public class GrpcClientStreamer implements Closeable {
try {
wait();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
if (isRunning()) {
@@ -181,6 +182,7 @@ public class GrpcClientStreamer implements Closeable {
try {
wait();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
if (!isRunning() && (!dataQueue.isEmpty() || !ackQueue.isEmpty())) {
@@ -200,6 +202,7 @@ public class GrpcClientStreamer implements Closeable {
try {
senderThread.join();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
proxyMap.close();
}
@@ -219,6 +222,7 @@ public class GrpcClientStreamer implements Closeable {
try {
GrpcClientStreamer.this.wait();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
if (running == RunningState.RUNNING) {
@@ -359,6 +363,7 @@ public class GrpcClientStreamer implements Closeable {
try {
exceptionAndRetry.retryInterval.sleep();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
leaderProxy.onNext(request);
}
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 8fbcc25..62ba7c9 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
@@ -164,8 +164,9 @@ public class GrpcLogAppender extends LogAppender {
try {
LOG.trace("{}: wait {}ms", this, waitTimeMs);
wait(waitTimeMs);
- } catch(InterruptedException ie) {
+ } catch (InterruptedException ie) {
LOG.warn(this + ": Wait interrupted by " + ie);
+ Thread.currentThread().interrupt();
}
}
}
@@ -278,7 +279,7 @@ public class GrpcLogAppender extends LogAppender {
try {
onNextImpl(reply);
- } catch(Throwable t) {
+ } catch(Exception t) {
LOG.error("Failed onNext request=" + request
+ ", reply=" + ServerProtoUtils.toString(reply), t);
}
@@ -498,6 +499,7 @@ public class GrpcLogAppender extends LogAppender {
try {
wait();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
}
@@ -542,6 +544,7 @@ public class GrpcLogAppender extends LogAppender {
try {
wait();
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
}
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java
index 21a6c56..7b3b62f 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolService.java
@@ -109,7 +109,7 @@ class GrpcServerProtocolService extends RaftServerProtocolServiceImplBase {
if (!replyInOrder(request)) {
try {
process(request).thenAccept(this::handleReply);
- } catch (Throwable e) {
+ } catch (Exception e) {
handleError(e, request);
}
return;
@@ -126,7 +126,7 @@ class GrpcServerProtocolService extends RaftServerProtocolServiceImplBase {
current.getFuture().complete(null);
return null;
});
- } catch (Throwable e) {
+ } catch (Exception e) {
handleError(e, request);
current.getFuture().completeExceptionally(e);
}
@@ -170,7 +170,7 @@ class GrpcServerProtocolService extends RaftServerProtocolServiceImplBase {
final RequestVoteReplyProto reply = server.requestVote(request);
responseObserver.onNext(reply);
responseObserver.onCompleted();
- } catch (Throwable e) {
+ } catch (Exception e) {
GrpcUtil.warn(LOG, () -> getId() + ": Failed requestVote " + ProtoUtils.toString(request.getServerRequest()), e);
responseObserver.onError(GrpcUtil.wrapException(e));
}
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcService.java
index a352bc1..4fb3a39 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcService.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcService.java
@@ -185,7 +185,8 @@ public final class GrpcService extends RaftServerRpcWithProxy<GrpcServerProtocol
super.closeImpl();
try {
s.awaitTermination();
- } catch(InterruptedException e) {
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException(name + " failed", e);
}
LOG.info("{} successfully", name);
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 45ac3d0..f80526b 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
@@ -682,6 +682,7 @@ public class LogStateMachine extends BaseStateMachine {
// avoid causing problem during leader election storm
Thread.sleep(10000);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
sendArchiveLogrequestToNewLeader(writer.getLastWrittenRecordId(), logName, location);
}
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java
index 5f5d7ec..52f9915 100644
--- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java
+++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java
@@ -62,6 +62,7 @@ public class NettyRpcProxy implements Closeable {
try {
return new NettyRpcProxy(peer, properties, group);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException("Failed connecting to " + peer, e);
}
}
@@ -183,6 +184,7 @@ public class NettyRpcProxy implements Closeable {
channelFuture.sync();
return reply.get(requestTimeoutDuration.getDuration(), requestTimeoutDuration.getUnit());
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException(ProtoUtils.toString(request)
+ " sending from " + peer + " is interrupted.", e);
} catch (ExecutionException e) {
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java
index 63e9301..220f9f0 100644
--- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java
+++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java
@@ -136,7 +136,7 @@ public final class NettyRpcService extends RaftServerRpcWithProxy<NettyRpcProxy,
public void startImpl() throws IOException {
try {
channelFuture.syncUninterruptibly();
- } catch(Throwable t) {
+ } catch(Exception t) {
throw new IOException(getId() + ": Failed to start " + getClass().getSimpleName(), t);
}
}
@@ -152,6 +152,7 @@ public final class NettyRpcService extends RaftServerRpcWithProxy<NettyRpcProxy,
workerGroup.awaitTermination(1000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.error("Interrupt EventLoopGroup terminate", e);
+ Thread.currentThread().interrupt();
}
super.closeImpl();
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java
index b1d0d93..2760b4a 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java
@@ -134,11 +134,12 @@ class FollowerState extends Daemon {
}
}
} catch (InterruptedException e) {
- LOG.info(this + " was interrupted: " + e);
+ LOG.info("{} was interrupted: {}", this, e);
LOG.trace("TRACE", e);
+ Thread.currentThread().interrupt();
return;
} catch (Exception e) {
- LOG.warn(this + " caught an exception", e);
+ LOG.warn("{} caught an exception", this, e);
}
}
}
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 5c023c6..039ce1d 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
@@ -163,7 +163,7 @@ class LeaderElection implements Runnable {
server.getLeaderElectionMetrics().getLeaderElectionTimer().time();
try {
askForVotes();
- } catch(Throwable e) {
+ } catch(Exception e) {
final LifeCycle.State state = lifeCycle.getCurrentState();
if (state.isClosingOrClosed()) {
LOG.info("{}: {} is safely ignored since this is already {}",
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 58fbc08..fc472dc 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
@@ -127,6 +127,7 @@ public class LeaderState {
queue.put(event);
} catch (InterruptedException e) {
LOG.info("{}: Interrupted when submitting {} ", this, event);
+ Thread.currentThread().interrupt();
}
}
@@ -134,7 +135,8 @@ public class LeaderState {
final StateUpdateEvent e;
try {
e = queue.poll(server.getMaxTimeoutMs(), TimeUnit.MILLISECONDS);
- } catch(InterruptedException ie) {
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
String s = this + ": poll() is interrupted";
if (!running) {
LOG.info(s + " gracefully");
@@ -737,6 +739,7 @@ public class LeaderState {
// leave some time for all RPC senders to send out new conf entry
Thread.sleep(server.getMinTimeoutMs());
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
// the pending request handler will send NotLeaderException for
// pending client requests when it stops
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 1fa5919..89ea15e 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
@@ -77,7 +77,10 @@ public class LogAppender {
}
try {
runAppenderImpl();
- } catch (InterruptedException | InterruptedIOException e) {
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.info(this + " was interrupted: " + e);
+ } catch (InterruptedIOException e) {
LOG.info(this + " was interrupted: " + e);
} catch (RaftLogIOException e) {
LOG.error(this + " failed RaftLog", e);
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 7665bd2..2277279 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
@@ -790,6 +790,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
} catch (InterruptedException e) {
final String s = id + ": Interrupted when waiting for reply, request=" + request;
LOG.info(s, e);
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException(s, e);
} catch (ExecutionException e) {
final Throwable cause = e.getCause();
@@ -1009,7 +1010,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
try {
return appendEntriesAsync(requestorId, r.getLeaderTerm(), previous, r.getLeaderCommit(),
request.getCallId(), r.getInitializing(), r.getCommitInfosList(), entries);
- } catch(Throwable t) {
+ } catch(Exception t) {
LOG.error("{}: Failed appendEntriesAsync {}", getMemberId(), r, t);
throw t;
}
@@ -1206,9 +1207,9 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
final InstallSnapshotReplyProto reply;
try {
reply = installSnapshotImpl(request);
- } catch (Throwable t) {
- LOG.error("{}: installSnapshot failed", getMemberId(), t);
- throw t;
+ } catch (Exception e) {
+ LOG.error("{}: installSnapshot failed", getMemberId(), e);
+ throw e;
}
if (LOG.isInfoEnabled()) {
LOG.info("{}: reply installSnapshot: {}", getMemberId(), ServerProtoUtils.toString(reply));
@@ -1536,7 +1537,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
CompletableFuture<Message> stateMachineFuture =
stateMachine.applyTransaction(trx);
return replyPendingRequest(next, stateMachineFuture);
- } catch (Throwable e) {
+ } catch (Exception e) {
LOG.error("{}: applyTransaction failed for index:{} proto:{}",
getMemberId(), next.getIndex(), ServerProtoUtils.toString(next), e);
throw e;
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 ab690d8..b1332e1 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
@@ -207,9 +207,9 @@ public class RaftServerProxy implements RaftServer {
if (!raftGroupId.filter(groupId::equals).isPresent()) {
addGroup(RaftGroup.valueOf(groupId));
}
- } catch (Throwable t) {
+ } catch (Exception e) {
LOG.warn(getId() + ": Failed to initialize the group directory "
- + sub.getAbsolutePath() + ". Ignoring it", t);
+ + sub.getAbsolutePath() + ". Ignoring it", e);
}
}));
raftGroup.ifPresent(this::addGroup);
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 24e05cd..f70ebc1 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
@@ -56,6 +56,7 @@ public final class ServerImplUtils {
() -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
5, sleepTime, "new RaftServerProxy", RaftServerProxy.LOG);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException(
"Interrupted when creating RaftServer " + id, e);
}
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 1905cef..9f0ddbc 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
@@ -400,6 +400,7 @@ public class ServerState implements Closeable {
try {
stateMachineUpdater.stopAndJoin();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
LOG.warn("{}: Interrupted when joining stateMachineUpdater", getMemberId(), e);
}
LOG.info("{}: closes. applyIndex: {}", getMemberId(), getLastAppliedIndex());
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 69dff24..60ee038 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
@@ -513,10 +513,10 @@ public abstract class RaftLog implements RaftLogSequentialOps, Closeable {
final String err = getName() + ": Timeout readStateMachineData for " + toLogEntryString(logEntry);
LOG.error(err, t);
throw t;
- } catch (Throwable t) {
+ } catch (Exception e) {
final String err = getName() + ": Failed readStateMachineData for " + toLogEntryString(logEntry);
- LOG.error(err, t);
- throw new RaftLogIOException(err, JavaUtils.unwrapCompletionException(t));
+ LOG.error(err, e);
+ throw new RaftLogIOException(err, JavaUtils.unwrapCompletionException(e));
}
// by this time we have already read the state machine data,
// so the log entry data should be set now
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java
index 27e733e..b6ab4b3 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java
@@ -308,7 +308,7 @@ public class SegmentedRaftLog extends RaftLog {
});
}
return new EntryWithData(entry, future);
- } catch (Throwable e) {
+ } catch (Exception e) {
final String err = getName() + ": Failed readStateMachineData for " +
ServerProtoUtils.toLogEntryString(entry);
LOG.error(err, e);
@@ -419,9 +419,9 @@ public class SegmentedRaftLog extends RaftLog {
cache.appendEntry(entry);
}
return writeFuture;
- } catch (Throwable throwable) {
- LOG.error("{}: Failed to append {}", getName(), ServerProtoUtils.toLogEntryString(entry), throwable);
- throw throwable;
+ } catch (Exception e) {
+ LOG.error("{}: Failed to append {}", getName(), ServerProtoUtils.toLogEntryString(entry), e);
+ throw e;
} finally {
context.stop();
}
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogInputStream.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogInputStream.java
index 1db790f..6d7e3b7 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogInputStream.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogInputStream.java
@@ -118,7 +118,7 @@ public class SegmentedRaftLogInputStream implements Closeable {
if (state.isUnopened()) {
try {
init();
- } catch (Throwable e) {
+ } catch (Exception e) {
LOG.error("caught exception initializing " + this, e);
throw IOUtils.asIOException(e);
}
@@ -240,10 +240,10 @@ public class SegmentedRaftLogInputStream implements Closeable {
} else {
hitError = false;
}
- } catch (Throwable t) {
+ } catch (Exception e) {
LOG.warn("Caught exception after scanning through {} ops from {}"
+ " while determining its valid length. Position was "
- + lastPos, numValid, in, t);
+ + lastPos, numValid, in, e);
hitError = true;
continue;
}
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 4a1e17d..563a11f 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
@@ -212,7 +212,7 @@ class SegmentedRaftLogReader implements Closeable {
in.reset();
throw e;
- } catch (Throwable e) {
+ } catch (Exception e) {
// raft log requires no gap between any two entries. thus if an entry is
// broken, throw the exception instead of skipping broken entries
in.reset();
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 402a222..42f2ab1 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
@@ -229,6 +229,7 @@ class SegmentedRaftLogWorker implements Runnable {
try {
workerThread.join(3000);
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
IOUtils.cleanup(LOG, out);
LOG.info("{} close()", name);
@@ -264,12 +265,12 @@ class SegmentedRaftLogWorker implements Runnable {
}
enqueueTimerContext.stop();
task.startTimerOnEnqueue(raftLogQueueingTimer);
- } catch (Throwable t) {
- if (t instanceof InterruptedException && !running) {
+ } catch (Exception e) {
+ if (e instanceof InterruptedException && !running) {
LOG.info("Got InterruptedException when adding task " + task
+ ". The SegmentedRaftLogWorker already stopped.");
} else {
- LOG.error("Failed to add IO task {}", task, t);
+ LOG.error("Failed to add IO task {}", task, e);
if (server != null) {
server.shutdown();
}
@@ -320,6 +321,7 @@ class SegmentedRaftLogWorker implements Runnable {
task.done();
}
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
if (running) {
LOG.warn("{} got interrupted while still running",
Thread.currentThread().getName());
@@ -327,14 +329,13 @@ class SegmentedRaftLogWorker implements Runnable {
LOG.info(Thread.currentThread().getName()
+ " was interrupted, exiting. There are " + queue.getNumElements()
+ " tasks remaining in the queue.");
- Thread.currentThread().interrupt();
return;
- } catch (Throwable t) {
+ } catch (Exception e) {
if (!running) {
LOG.info("{} got closed and hit exception",
- Thread.currentThread().getName(), t);
+ Thread.currentThread().getName(), e);
} else {
- LOG.error("{} hit exception", Thread.currentThread().getName(), t);
+ LOG.error("{} hit exception", Thread.currentThread().getName(), e);
// Shutdown raft group instead of terminating jvm.
if (server != null) {
server.shutdown();
@@ -465,7 +466,7 @@ class SegmentedRaftLogWorker implements Runnable {
try {
// this.entry != entry iff the entry has state machine data
this.stateMachineFuture = stateMachine.data().write(entry);
- } catch (Throwable e) {
+ } catch (Exception e) {
LOG.error(name + ": writeStateMachineData failed for index " + entry.getIndex()
+ ", entry=" + ServerProtoUtils.toLogEntryString(entry, stateMachine::toStateMachineLogEntryString), e);
throw e;
diff --git a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java
index fb6050a..6c27ffe 100644
--- a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java
+++ b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java
@@ -123,7 +123,7 @@ public abstract class MiniRaftCluster implements Closeable {
cluster.start();
}
testCase.accept(cluster);
- } catch(Throwable t) {
+ } catch(Exception t) {
LOG.info(cluster.printServers());
LOG.error("Failed " + caller, t);
throw t;
@@ -139,7 +139,7 @@ public abstract class MiniRaftCluster implements Closeable {
try {
cluster = getFactory().reuseCluster(numServers, getProperties());
testCase.accept(cluster);
- } catch(Throwable t) {
+ } catch(Exception t) {
if (cluster != null) {
LOG.info(cluster.printServers());
}
@@ -725,8 +725,9 @@ public abstract class MiniRaftCluster implements Closeable {
executor.shutdown();
// just wait for a few seconds
executor.awaitTermination(5, TimeUnit.SECONDS);
- } catch(InterruptedException e) {
+ } catch (InterruptedException e) {
LOG.warn("shutdown interrupted", e);
+ Thread.currentThread().interrupt();
}
Optional.ofNullable(timer.get()).ifPresent(Timer::cancel);
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
index 42fd159..6d740ff 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
@@ -314,7 +314,7 @@ public abstract class RaftBasicTests<CLUSTER extends MiniRaftCluster>
f.join();
Assert.assertTrue(step.get() == messages.length);
}
- } catch(Throwable t) {
+ } catch(Exception t) {
if (exceptionInClientThread.compareAndSet(null, t)) {
LOG.error(this + " failed", t);
} else {
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftTestUtil.java b/ratis-server/src/test/java/org/apache/ratis/RaftTestUtil.java
index 05435ca..1a8a1fb 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftTestUtil.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftTestUtil.java
@@ -275,7 +275,7 @@ public interface RaftTestUtil {
final List<LogEntryProto> entries = getStateMachineLogEntries(log);
try {
assertLogEntries(entries, expectedTerm, expectedMessages);
- } catch(Throwable t) {
+ } catch(Exception t) {
throw new AssertionError("entries: " + entries, t);
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/BlockRequestHandlingInjection.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/BlockRequestHandlingInjection.java
index abe9dd5..125641f 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/impl/BlockRequestHandlingInjection.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/BlockRequestHandlingInjection.java
@@ -78,6 +78,7 @@ public class BlockRequestHandlingInjection implements CodeInjectionForTesting.Co
RaftTestUtil.block(() -> shouldBlock(localId, remoteId));
} catch (InterruptedException e) {
LOG.debug("Interrupted while blocking request from " + remoteId + " to " + localId, e);
+ Thread.currentThread().interrupt();
}
LOG.info(localId + ": unBlock request from " + remoteId);
return true;
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/DelayLocalExecutionInjection.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/DelayLocalExecutionInjection.java
index 410228f..0d7d79e 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/impl/DelayLocalExecutionInjection.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/DelayLocalExecutionInjection.java
@@ -67,6 +67,7 @@ public class DelayLocalExecutionInjection implements CodeInjectionForTesting.Cod
RaftTestUtil.delay(d::get);
} catch (InterruptedException e) {
LOG.debug("Interrupted while delaying " + localIdStr);
+ Thread.currentThread().interrupt();
}
return true;
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java
index b54fbdc..3675137 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java
@@ -679,6 +679,7 @@ public abstract class RaftReconfigurationBaseTest<CLUSTER extends MiniRaftCluste
try {
Thread.sleep(200);
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/StateMachineShutdownTests.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/StateMachineShutdownTests.java
index a6ffb30..758cd88 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/impl/StateMachineShutdownTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/StateMachineShutdownTests.java
@@ -52,6 +52,7 @@ public abstract class StateMachineShutdownTests<CLUSTER extends MiniRaftCluster>
try {
objectToWait.wait();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new RuntimeException();
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/simulation/RequestHandler.java b/ratis-server/src/test/java/org/apache/ratis/server/simulation/RequestHandler.java
index 802797f..10382f0 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/simulation/RequestHandler.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/simulation/RequestHandler.java
@@ -129,12 +129,12 @@ public class RequestHandler<REQUEST extends RaftRpcMessage,
} catch (IOException e) {
LOG.error(this + " has " + e);
LOG.trace("TRACE", e);
- } catch(Throwable t) {
+ } catch(Exception e) {
if (!handlerImpl.isAlive()) {
LOG.info(this + " is stopped.");
break;
}
- ExitUtils.terminate(1, this + " is terminating.", t, LOG);
+ ExitUtils.terminate(1, this + " is terminating.", e, LOG);
}
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedRequestReply.java b/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedRequestReply.java
index 572ca61..9e4299d 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedRequestReply.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedRequestReply.java
@@ -120,6 +120,7 @@ class SimulatedRequestReply<REQUEST extends RaftRpcMessage, REPLY extends RaftRp
RaftTestUtil.block(q.blockSendRequestTo::get);
return q.request(request);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException("", e);
}
}
@@ -149,6 +150,7 @@ class SimulatedRequestReply<REQUEST extends RaftRpcMessage, REPLY extends RaftRp
break;
}
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException("", e);
}
return request;
@@ -191,6 +193,7 @@ class SimulatedRequestReply<REQUEST extends RaftRpcMessage, REPLY extends RaftRp
try {
Thread.sleep(randomSleepMs);
} catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
throw IOUtils.toInterruptedIOException("", ie);
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedServerRpc.java b/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedServerRpc.java
index 05f47f5..507d04d 100644
--- a/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedServerRpc.java
+++ b/ratis-server/src/test/java/org/apache/ratis/server/simulation/SimulatedServerRpc.java
@@ -90,6 +90,7 @@ class SimulatedServerRpc implements RaftServerRpc {
executor.shutdown();
executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
} catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
clientHandler.shutdown();
serverHandler.shutdown();
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 6fd9712..5653bd7 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
@@ -152,7 +152,10 @@ public class SimpleStateMachine4Testing extends BaseStateMachine {
void await(Type type) {
try {
getFuture(type).get();
- } catch(InterruptedException | ExecutionException e) {
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Failed to await " + type, e);
+ } catch(ExecutionException e) {
throw new IllegalStateException("Failed to await " + type, e);
}
}
@@ -175,7 +178,8 @@ public class SimpleStateMachine4Testing extends BaseStateMachine {
try {
TimeUnit.SECONDS.sleep(1);
- } catch(InterruptedException ignored) {
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
}
}
});
diff --git a/ratis-test/src/test/java/org/apache/ratis/retry/TestExponentialBackoffRetry.java b/ratis-test/src/test/java/org/apache/ratis/retry/TestExponentialBackoffRetry.java
index 232077e..74f1807 100644
--- a/ratis-test/src/test/java/org/apache/ratis/retry/TestExponentialBackoffRetry.java
+++ b/ratis-test/src/test/java/org/apache/ratis/retry/TestExponentialBackoffRetry.java
@@ -42,7 +42,7 @@ public class TestExponentialBackoffRetry extends BaseTest {
// baseSleep should not be null
createPolicy(null, null, 1);
Assert.fail("Policy creation should have failed");
- } catch (Throwable t) {
+ } catch (Exception e) {
}
// test policy without max sleep
diff --git a/ratis-test/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java b/ratis-test/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java
index 180f970..2965a2d 100644
--- a/ratis-test/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java
+++ b/ratis-test/src/test/java/org/apache/ratis/statemachine/TestStateMachine.java
@@ -103,8 +103,8 @@ public class TestStateMachine extends BaseTest implements MiniRaftClusterWithSim
assertNull(context);
}
numApplied.incrementAndGet();
- } catch (Throwable t) {
- throwable.set(t);
+ } catch (Exception e) {
+ throwable.set(e);
}
return CompletableFuture.completedFuture(null);
}
diff --git a/ratis-test/src/test/java/org/apache/ratis/util/TestDataBlockingQueue.java b/ratis-test/src/test/java/org/apache/ratis/util/TestDataBlockingQueue.java
index cfd45d5..7017f97 100644
--- a/ratis-test/src/test/java/org/apache/ratis/util/TestDataBlockingQueue.java
+++ b/ratis-test/src/test/java/org/apache/ratis/util/TestDataBlockingQueue.java
@@ -83,8 +83,8 @@ public class TestDataBlockingQueue {
}
assertOfferPull(offeringValue.get(), polledValue.get(), elementLimit);
}
- } catch (Throwable t) {
- ExitUtils.terminate(-2, "pollThread failed", t, null);
+ } catch (Exception e) {
+ ExitUtils.terminate(-2, "pollThread failed", e, null);
}
});
@@ -98,8 +98,8 @@ public class TestDataBlockingQueue {
}
assertOfferPull(offeringValue.get(), polledValue.get(), elementLimit);
}
- } catch (Throwable t) {
- ExitUtils.terminate(-1, "offerThread failed", t, null);
+ } catch (Exception e) {
+ ExitUtils.terminate(-1, "offerThread failed", e, null);
}
});
diff --git a/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java b/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
index dfed847..d085161 100644
--- a/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
+++ b/ratis-test/src/test/java/org/apache/ratis/util/TestResourceSemaphore.java
@@ -97,6 +97,7 @@ public class TestResourceSemaphore extends BaseTest {
try {
g.acquire(permits);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
e.printStackTrace();
}
};