Repository: apache/mina-sshd
Issue: requestNewKeysExchange may cause KeyExchangeMessageHandler into dead lock,cpu 100%

### Version

2.9.1

### Bug description

i observed in cases， the write packet thread is always running  and in the interval i saw "equestNewKeysExchange(ClientSessionImpl[root@/47.104.92.124:22]) Initiating key re-exchange
method:o.a.s.c.s.ClientSessionImpl(AbstractSession.java:2404) " in log 

### Actual behavior

after i saw log "equestNewKeysExchange(ClientSessionImpl[root@/47.104.92.124:22]) Initiating key re-exchange
method:o.a.s.c.s.ClientSessionImpl(AbstractSession.java:2404)",the write packet thread is soar to 100%,not stopped

### Expected behavior

i expect when the connection is closed the cpu is down,and the thread is stopped

### Relevant log output

```Shell
equestNewKeysExchange(ClientSessionImpl[root@/47.104.92.124:22]) Initiating key re-exchange
method:o.a.s.c.s.ClientSessionImpl(AbstractSession.java:2404)
```


### Other information

_No response_

Changed files:
sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java

diff --git a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
index 294935a0a..11501ef77 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
@@ -34,6 +34,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Supplier;
 
 import org.apache.sshd.common.SshConstants;
+import org.apache.sshd.common.SshException;
 import org.apache.sshd.common.future.DefaultKeyExchangeFuture;
 import org.apache.sshd.common.io.IoWriteFuture;
 import org.apache.sshd.common.kex.KexState;
@@ -104,6 +105,11 @@ public class KeyExchangeMessageHandler {
      */
     protected volatile boolean kexFlushed = true;
 
+    /**
+     * Indicates that the handler has been shut down.
+     */
+    protected volatile boolean shutDown;
+
     /**
      * Never {@code null}. Used to block some threads when writing packets while pending packets are still being flushed
      * at the end of a KEX to avoid overrunning the flushing thread. Always set, initially fulfilled. At the beginning
@@ -194,6 +200,7 @@ public class KeyExchangeMessageHandler {
     public void shutdown() {
         SimpleImmutableEntry<Integer, DefaultKeyExchangeFuture> items = updateState(() -> {
             kexFlushed = true;
+            shutDown = true;
             return new SimpleImmutableEntry<Integer, DefaultKeyExchangeFuture>(
                     Integer.valueOf(pendingPackets.size()),
                     kexFlushedFuture);
@@ -229,33 +236,33 @@ public class KeyExchangeMessageHandler {
         boolean enqueued = false;
         boolean isLowLevelMessage = cmd <= SshConstants.SSH_MSG_KEX_LAST && cmd != SshConstants.SSH_MSG_SERVICE_REQUEST
                 && cmd != SshConstants.SSH_MSG_SERVICE_ACCEPT;
+        IoWriteFuture future = null;
         try {
             if (isLowLevelMessage) {
                 // Low-level messages can always be sent.
-                return session.doWritePacket(buffer);
+                future = session.doWritePacket(buffer);
+            } else {
+                future = writeOrEnqueue(cmd, buffer, timeout, unit);
+                enqueued = future instanceof PendingWriteFuture;
             }
-            IoWriteFuture future = writeOrEnqueue(cmd, buffer, timeout, unit);
-            enqueued = future instanceof PendingWriteFuture;
-            return future;
         } finally {
             session.resetIdleTimeout();
-            if (!enqueued) {
-                try {
-                    session.checkRekey();
-                } catch (GeneralSecurityException e) {
-                    if (log.isDebugEnabled()) {
-                        log.debug("writePacket({}) failed ({}) to check re-key: {}", session, e.getClass().getSimpleName(),
-                                e.getMessage(), e);
-                    }
-                    throw ValidateUtils.initializeExceptionCause(
-                            new ProtocolException("Failed (" + e.getClass().getSimpleName() + ")"
-                                                  + " to check re-key necessity: " + e.getMessage()),
-                            e);
-                } catch (Exception e) {
-                    ExceptionUtils.rethrowAsIoException(e);
+        }
+        if (!enqueued) {
+            try {
+                session.checkRekey();
+            } catch (GeneralSecurityException e) {
+                if (log.isDebugEnabled()) {
+                    log.debug("writePacket({}) failed ({}) to check re-key: {}", session, e.getClass().getSimpleName(),
+                            e.getMessage(), e);
                 }
+                throw ValidateUtils.initializeExceptionCause(new ProtocolException(
+                        "Failed (" + e.getClass().getSimpleName() + ")" + " to check re-key necessity: " + e.getMessage()), e);
+            } catch (Exception e) {
+                ExceptionUtils.rethrowAsIoException(e);
             }
         }
+        return future;
     }
 
     /**
@@ -288,6 +295,9 @@ public class KeyExchangeMessageHandler {
             // Use the readLock here to give KEX state updates and the flushing thread priority.
             lock.readLock().lock();
             try {
+                if (shutDown) {
+                    throw new SshException("Write attempt on closing session: " + SshConstants.getCommandMessageName(cmd));
+                }
                 KexState state = session.kexState.get();
                 boolean kexDone = KexState.DONE.equals(state) || KexState.KEYS.equals(state);
                 if (kexDone && kexFlushed) {