STAR-1690 Stop message processing after MessagingService shutdown (apache#581)

* STAR-1690 Stop message processing after MessagingService shutdown
  The RequestCallbacks class now stops adding callbacks after it
  is shutdown by the MessagingService when that is shutdown.

* Make new log statements trace

Author: mike-tr-adamson

src/java/org/apache/cassandra/net/RequestCallbacks.java

@@ -67,6 +67,8 @@ public class RequestCallbacks implements OutboundMessageCallbacks
     private final ScheduledExecutorService executor = new DebuggableScheduledThreadPoolExecutor("Callback-Map-Reaper");
     private final ConcurrentMap<CallbackKey, CallbackInfo> callbacks = new ConcurrentHashMap<>();
 
+    private volatile boolean shutdown;
+
     RequestCallbacks(MessagingService messagingService)
     {
         this.messagingService = messagingService;
@@ -100,6 +102,12 @@ void addWithExpiration(RequestCallback cb, Message message, InetAddressAndPort t
     {
         // mutations need to call the overload with a ConsistencyLevel
         assert message.verb() != Verb.MUTATION_REQ && message.verb() != Verb.COUNTER_MUTATION_REQ && message.verb() != Verb.PAXOS_COMMIT_REQ;
+        if (shutdown)
+        {
+            if (logger.isTraceEnabled())
+                logger.trace("Received request after messaging service shutdown so ignoring it");
+            return;
+        }
         CallbackInfo previous = callbacks.put(key(message.id(), to), new CallbackInfo(message, to, cb));
         assert previous == null : format("Callback already exists for id %d/%s! (%s)", message.id(), to, previous);
     }
@@ -112,6 +120,12 @@ public void addWithExpiration(AbstractWriteResponseHandler<?> cb,
                                   boolean allowHints)
     {
         assert message.verb() == Verb.MUTATION_REQ || message.verb() == Verb.COUNTER_MUTATION_REQ || message.verb() == Verb.PAXOS_COMMIT_REQ;
+        if (shutdown)
+        {
+            if (logger.isTraceEnabled())
+                logger.trace("Received request after messaging service shutdown so ignoring it");
+            return;
+        }
         CallbackInfo previous = callbacks.put(key(message.id(), to.endpoint()), new WriteCallbackInfo(message, to, cb, consistencyLevel, allowHints));
         assert previous == null : format("Callback already exists for id %d/%s! (%s)", message.id(), to.endpoint(), previous);
     }
@@ -181,13 +195,15 @@ private void onExpired(CallbackInfo info)
 
     void shutdownNow(boolean expireCallbacks)
     {
+        shutdown = true;
         executor.shutdownNow();
         if (expireCallbacks)
             forceExpire();
     }
 
     void shutdownGracefully()
     {
+        shutdown = true;
         expire();
         if (!callbacks.isEmpty())
             executor.schedule(this::shutdownGracefully, 100L, MILLISECONDS);