RATIS-1709 Support specify ThreadGroup for Daemon threads

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }

ratis-common/src/main/java/org/apache/ratis/util/Daemon.java

@@ -26,7 +26,7 @@ public class Daemon extends Thread {
 
  /** Construct a daemon thread with flexible arguments. */
  protected Daemon(Builder builder) {
- super(builder.runnable);
+ super(builder.threadGroup, builder.runnable);
  setName(builder.name);
  }
 
@@ -38,6 +38,7 @@ public static Builder newBuilder() {
  public static class Builder {
  private String name;
  private Runnable runnable;
+ private ThreadGroup threadGroup;
 
  public Builder setName(String name) {
  this.name = name;
@@ -49,6 +50,11 @@ public Builder setRunnable(Runnable runnable) {
  return this;
  }
 
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
+
  public Daemon build() {
  Objects.requireNonNull(name, "name == null");
  return new Daemon(this);

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java

@@ -116,6 +116,9 @@ default RaftGroup getGroup() {
  /** @return the internal {@link RaftClient} of this division. */
  RaftClient getRaftClient();
 
+ /** @return the {@link ThreadGroup} the threads of this Division belong to. */
+ ThreadGroup getThreadGroup();
+
  @Override
  void close();
  }
@@ -168,7 +171,7 @@ class Builder {
  private static Method initNewRaftServerMethod() {
  final String className = RaftServer.class.getPackage().getName() + ".impl.ServerImplUtils";
  final Class<?>[] argClasses = {RaftPeerId.class, RaftGroup.class, RaftStorage.StartupOption.class,
- StateMachine.Registry.class, RaftProperties.class, Parameters.class};
+ StateMachine.Registry.class, ThreadGroup.class, RaftProperties.class, Parameters.class};
  try {
  final Class<?> clazz = ReflectionUtils.getClassByName(className);
  return clazz.getMethod("newRaftServer", argClasses);
@@ -178,11 +181,11 @@ private static Method initNewRaftServerMethod() {
  }
 
  private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, RaftStorage.StartupOption option,
- StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  try {
  return (RaftServer) NEW_RAFT_SERVER_METHOD.invoke(null,
- serverId, group, option, stateMachineRegistry, properties, parameters);
+ serverId, group, option, stateMachineRegistry, threadGroup, properties, parameters);
  } catch (IllegalAccessException e) {
  throw new IllegalStateException("Failed to build " + serverId, e);
  } catch (InvocationTargetException e) {
@@ -196,6 +199,7 @@ private static RaftServer newRaftServer(RaftPeerId serverId, RaftGroup group, Ra
  private RaftStorage.StartupOption option = RaftStorage.StartupOption.FORMAT;
  private RaftProperties properties;
  private Parameters parameters;
+ private ThreadGroup threadGroup;
 
  /** @return a {@link RaftServer} object. */
  public RaftServer build() throws IOException {
@@ -205,6 +209,7 @@ public RaftServer build() throws IOException {
  option,
  Objects.requireNonNull(stateMachineRegistry , "Neither 'stateMachine' nor 'setStateMachineRegistry' " +
  "is initialized."),
+ threadGroup,
  Objects.requireNonNull(properties, "The 'properties' field is not initialized."),
  parameters);
  }
@@ -249,5 +254,15 @@ public Builder setParameters(Parameters parameters) {
  this.parameters = parameters;
  return this;
  }
+
+ /**
+ * Set {@link ThreadGroup} so the application can control RaftServer threads consistently with the application.
+ * For example, configure {@link ThreadGroup#uncaughtException(Thread, Throwable)} for the whole thread group.
+ * If not set, the new thread will be put into the thread group of the caller thread.
+ */
+ public Builder setThreadGroup(ThreadGroup threadGroup) {
+ this.threadGroup = threadGroup;
+ return this;
+ }
  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -64,7 +64,9 @@ int update(AtomicInteger outstanding) {
  private final AtomicInteger outstandingOp = new AtomicInteger();
 
  FollowerState(RaftServerImpl server, Object reason) {
- super(newBuilder().setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class)));
+ super(newBuilder()
+ .setName(server.getMemberId() + "-" + JavaUtils.getClassSimpleName(FollowerState.class))
+ .setThreadGroup(server.getThreadGroup()));
  this.server = server;
  this.reason = reason;
  }

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java

@@ -190,7 +190,8 @@ public String toString() {
  LeaderElection(RaftServerImpl server, boolean skipPreVote) {
  this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass()) + COUNT.incrementAndGet();
  this.lifeCycle = new LifeCycle(this);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.server = server;
  this.skipPreVote = skipPreVote ||
  !RaftServerConfigKeys.LeaderElection.preVote(

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -272,7 +272,7 @@ boolean removeAll(Collection<LogAppender> c) {
  this.currentTerm = state.getCurrentTerm();
 
  this.eventQueue = new EventQueue();
- processor = new EventProcessor(this.name);
+ processor = new EventProcessor(this.name, server);
  raftServerMetrics = server.getRaftServerMetrics();
  logAppenderMetrics = new LogAppenderMetrics(server.getMemberId());
  this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
@@ -620,8 +620,9 @@ private void prepare() {
  * state, such as changing to follower, or updating the committed index.
  */
  private class EventProcessor extends Daemon {
- public EventProcessor(String name) {
- super(Daemon.newBuilder().setName(name));
+ public EventProcessor(String name, RaftServerImpl server) {
+ super(Daemon.newBuilder()
+ .setName(name).setThreadGroup(server.getThreadGroup()));
  }
  @Override
  public void run() {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -97,7 +97,7 @@
 
 class RaftServerImpl implements RaftServer.Division,
  RaftServerProtocol, RaftServerAsynchronousProtocol,
- RaftClientProtocol, RaftClientAsynchronousProtocol{
+ RaftClientProtocol, RaftClientAsynchronousProtocol{
  private static final String CLASS_NAME = JavaUtils.getClassSimpleName(RaftServerImpl.class);
  static final String REQUEST_VOTE = CLASS_NAME + ".requestVote";
  static final String APPEND_ENTRIES = CLASS_NAME + ".appendEntries";
@@ -189,6 +189,7 @@ public long[] getFollowerNextIndices() {
  private final ExecutorService clientExecutor;
 
  private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
+ private final ThreadGroup threadGroup;
 
  RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
  throws IOException {
@@ -216,6 +217,7 @@ public long[] getFollowerNextIndices() {
  getMemberId(), () -> commitInfoCache::get, retryCache::getStatistics);
 
  this.startComplete = new AtomicBoolean(false);
+ this.threadGroup = new ThreadGroup(proxy.getThreadGroup(), getMemberId().toString());
 
  this.raftClient = JavaUtils.memoize(() -> RaftClient.newBuilder()
  .setRaftGroup(group)
@@ -274,6 +276,11 @@ TimeDuration getSleepDeviationThreshold() {
  return sleepDeviationThreshold;
  }
 
+ @Override
+ public ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public StateMachine getStateMachine() {
  return stateMachine;

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java

@@ -196,9 +196,10 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  private final ExecutorService executor;
 
  private final JvmPauseMonitor pauseMonitor;
+ private final ThreadGroup threadGroup;
 
  RaftServerProxy(RaftPeerId id, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) {
+ RaftProperties properties, Parameters parameters, ThreadGroup threadGroup) {
  this.properties = properties;
  this.stateMachineRegistry = stateMachineRegistry;
 
@@ -221,6 +222,7 @@ String toString(RaftGroupId groupId, CompletableFuture<RaftServerImpl> f) {
  final TimeDuration leaderStepDownWaitTime = RaftServerConfigKeys.LeaderElection.leaderStepDownWaitTime(properties);
  this.pauseMonitor = new JvmPauseMonitor(id,
  extraSleep -> handleJvmPause(extraSleep, rpcSlownessTimeout, leaderStepDownWaitTime));
+ this.threadGroup = threadGroup == null ? new ThreadGroup(this.id.toString()) : threadGroup;
  }
 
  private void handleJvmPause(TimeDuration extraSleep, TimeDuration closeThreshold, TimeDuration stepDownThreshold)
@@ -379,6 +381,10 @@ public LifeCycle.State getLifeCycleState() {
  return lifeCycle.getCurrentState();
  }
 
+ ThreadGroup getThreadGroup() {
+ return threadGroup;
+ }
+
  @Override
  public void start() throws IOException {
  lifeCycle.startAndTransition(this::startImpl, IOException.class);

ratis-server/src/main/java/org/apache/ratis/server/impl/ServerImplUtils.java

@@ -47,26 +47,26 @@ private ServerImplUtils() {
  /** Create a {@link RaftServerProxy}. */
  public static RaftServerProxy newRaftServer(
  RaftPeerId id, RaftGroup group, RaftStorage.StartupOption option, StateMachine.Registry stateMachineRegistry,
- RaftProperties properties, Parameters parameters) throws IOException {
+ThreadGroup threadGroup, RaftProperties properties, Parameters parameters) throws IOException {
  RaftServer.LOG.debug("newRaftServer: {}, {}", id, group);
  if (group != null && !group.getPeers().isEmpty()) {
  Preconditions.assertNotNull(id, "RaftPeerId %s is not in RaftGroup %s", id, group);
  Preconditions.assertNotNull(group.getPeer(id), "RaftPeerId %s is not in RaftGroup %s", id, group);
  }
- final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, properties, parameters);
+ final RaftServerProxy proxy = newRaftServer(id, stateMachineRegistry, threadGroup, properties, parameters);
  proxy.initGroups(group, option);
  return proxy;
  }
 
  private static RaftServerProxy newRaftServer(
- RaftPeerId id, StateMachine.Registry stateMachineRegistry, RaftProperties properties, Parameters parameters)
- throws IOException {
+ RaftPeerId id, StateMachine.Registry stateMachineRegistry, ThreadGroup threadGroup, RaftProperties properties,
+Parameters parameters) throws IOException {
  final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
  final RaftServerProxy proxy;
  try {
  // attempt multiple times to avoid temporary bind exception
  proxy = JavaUtils.attemptRepeatedly(
- () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters),
+ () -> new RaftServerProxy(id, stateMachineRegistry, properties, parameters, threadGroup),
  5, sleepTime, "new RaftServerProxy", RaftServer.LOG);
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();

ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java

@@ -113,8 +113,8 @@ public int getNumSnapshotsRetained() {
  }
  };
  this.purgeUptoSnapshotIndex = RaftServerConfigKeys.Log.purgeUptoSnapshotIndex(properties);
-
-updater = Daemon.newBuilder().setName(name).setRunnable(this).build();
+ updater = Daemon.newBuilder().setName(name).setRunnable(this)
+ .setThreadGroup(server.getThreadGroup()).build();
  this.awaitForSignal = new AwaitForSignal(name);
  this.stateMachineMetrics = MemoizedSupplier.valueOf(
  () -> StateMachineMetrics.getStateMachineMetrics(server, appliedIndex, stateMachine));

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDaemon.java

@@ -48,7 +48,8 @@ class LogAppenderDaemon {
  this.logAppender = logAppender;
  this.name = logAppender + "-" + JavaUtils.getClassSimpleName(getClass());
  this.lifeCycle = new LifeCycle(name);
- this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run).build();
+ this.daemon = Daemon.newBuilder().setName(name).setRunnable(this::run)
+ .setThreadGroup(logAppender.getServer().getThreadGroup()).build();
  }
 
  public boolean isWorking() {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -390,7 +390,7 @@ private RaftServerProxy newRaftServer(RaftPeerId id, RaftGroup group, boolean fo
  RaftServerConfigKeys.setStorageDir(prop, Collections.singletonList(dir));
  return ServerImplUtils.newRaftServer(id, group,
  format? RaftStorage.StartupOption.FORMAT: RaftStorage.StartupOption.RECOVER,
- getStateMachineRegistry(prop), prop, setPropertiesAndInitParameters(id, group, prop));
+ getStateMachineRegistry(prop), null, prop, setPropertiesAndInitParameters(id, group, prop));
  } catch (IOException e) {
  throw new RuntimeException(e);
  }