Update OOM handling outside of timeout

Author: iit2009060

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -899,6 +899,7 @@ protected Node curNode() {
          * @param now       The current time in milliseconds.
          * @param throwable The failure exception.
          */
+        @SuppressWarnings("NPathComplexity")
         final void fail(long now, Throwable throwable) {
             if (curNode != null) {
                 runnable.nodeReadyDeadlines.remove(curNode);
@@ -920,6 +921,11 @@ final void fail(long now, Throwable throwable) {
             }
             nextAllowedTryMs = now + retryBackoff.backoff(tries++);
 
+            // Don't mask OutOfMemoryError as TimeoutException - propagate it directly
+            if (throwable instanceof OutOfMemoryError) {
+                handleFailure(throwable);
+                return;
+            }
             // If the call has timed out, fail.
             if (calcTimeoutMsRemainingAsInt(now, deadlineMs) <= 0) {
                 handleTimeoutFailure(now, throwable);
@@ -955,10 +961,7 @@ private void handleTimeoutFailure(long now, Throwable cause) {
                 log.debug("{} timed out at {} after {} attempt(s)", this, now, tries,
                     new Exception(prettyPrintException(cause)));
             }
-            // Don't mask OutOfMemoryError as TimeoutException - propagate it directly
-            if (cause instanceof OutOfMemoryError) {
-                handleFailure(cause);
-            } else if (cause instanceof TimeoutException) {
+            if (cause instanceof TimeoutException) {
                 handleFailure(cause);
             } else {
                 handleFailure(new TimeoutException(this + " timed out at " + now