Address PR review feedback on async refresh handling

- Move refreshInProgress reset to a finally block so a thrown exception
  in updateToken or cachedTokenIsNewer cannot permanently deadlock future
  async refreshes.
- Update the memory-ordering comment in updateToken to acknowledge that
  handleFailedAsyncRefresh writes staleAfter without a subsequent volatile
  token write (consequence is at most one extra async trigger).
- Add a comment in triggerAsyncRefresh explaining why only STALE (not
  EXPIRED) triggers an async attempt.
- Add testAsyncRefreshDiscardsOlderToken to cover the cachedTokenIsNewer
  discard path where a blocking refresh installs a newer token while an
  async refresh is in flight.

Author: mihaimitrea-db

databricks-sdk-java/src/main/java/com/databricks/sdk/core/oauth/CachedTokenSource.java

@@ -221,7 +221,9 @@ private void updateToken(Token t) {
     }
 
     // Publish the token after staleAfter so readers that observe the new token also observe the
-    // stale threshold computed for that token.
+    // stale threshold computed for that token. Note: handleFailedAsyncRefresh writes staleAfter
+    // without a subsequent volatile token write, so a concurrent reader may briefly see a stale
+    // staleAfter value; the only consequence is one extra async trigger, which is harmless.
     this.token = t;
   }
 
@@ -353,8 +355,9 @@ protected Token getTokenAsync() {
    * succeeded.
    */
   private synchronized void triggerAsyncRefresh() {
-    // Check token state again inside the synchronized block to avoid triggering a refresh if
-    // another thread updated the token in the meantime.
+    // Re-check inside the synchronized block: another thread may have updated the token.
+    // Only STALE triggers async refresh; EXPIRED tokens are handled by getTokenBlocking, so
+    // an async attempt is unnecessary and would race with the blocking path.
     if (refreshInProgress || getTokenState(token) != TokenState.STALE) {
       return;
     }
@@ -368,14 +371,16 @@ private synchronized void triggerAsyncRefresh() {
               if (newToken != null && !cachedTokenIsNewer(newToken)) {
                 updateToken(newToken);
               }
-              refreshInProgress = false;
             }
           } catch (Exception e) {
             synchronized (this) {
               handleFailedAsyncRefresh();
-              refreshInProgress = false;
               logger.error("Asynchronous token refresh failed", e);
             }
+          } finally {
+            synchronized (this) {
+              refreshInProgress = false;
+            }
           }
         });
   }

databricks-sdk-java/src/test/java/com/databricks/sdk/core/oauth/CachedTokenSourceTest.java

@@ -317,6 +317,79 @@ void testGetTokenRetriesAfterAsyncBackoffElapsesAndUpdatesToken() throws Excepti
     assertEquals(2, refreshCallCount.get(), "The cache should retry exactly once after backoff");
   }
 
+  /**
+   * Verifies that an async refresh result is discarded when the cache already holds a token with a
+   * later expiry. This covers the concurrent scenario where a blocking refresh runs while an async
+   * refresh is in flight: the async result is older and should not overwrite the newer cached token.
+   */
+  @Test
+  void testAsyncRefreshDiscardsOlderToken() throws Exception {
+    TestClockSupplier clockSupplier = new TestClockSupplier(BASE_TIME);
+
+    Token olderRefreshToken =
+        tokenExpiringAt("older-async-token", BASE_TIME.plus(Duration.ofMinutes(8)));
+    Token newerBlockingToken =
+        tokenExpiringAt("newer-blocking-token", BASE_TIME.plus(Duration.ofMinutes(20)));
+
+    CountDownLatch asyncRefreshStarted = new CountDownLatch(1);
+    CountDownLatch allowAsyncToFinish = new CountDownLatch(1);
+    AtomicInteger refreshCallCount = new AtomicInteger();
+
+    TokenSource tokenSource =
+        () -> {
+          int call = refreshCallCount.incrementAndGet();
+          if (call == 1) {
+            asyncRefreshStarted.countDown();
+            try {
+              allowAsyncToFinish.await(5, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+            return olderRefreshToken;
+          }
+          return newerBlockingToken;
+        };
+
+    CachedTokenSource source =
+        new CachedTokenSource.Builder(tokenSource)
+            .setToken(tokenExpiringAt(INITIAL_TOKEN, BASE_TIME.plus(Duration.ofMinutes(10))))
+            .setClockSupplier(clockSupplier)
+            .build();
+
+    // Advance clock so the token is stale, triggering an async refresh.
+    clockSupplier.advanceTime(Duration.ofMinutes(6));
+    Token staleResult = source.getToken();
+    assertEquals(INITIAL_TOKEN, staleResult.getAccessToken());
+    assertTrue(
+        asyncRefreshStarted.await(1, TimeUnit.SECONDS),
+        "Async refresh should have started");
+
+    // While async refresh is blocked, advance time so the token expires and force a blocking
+    // refresh that installs a newer token.
+    clockSupplier.advanceTime(Duration.ofMinutes(4));
+    Token blockingResult = source.getToken();
+    assertEquals("newer-blocking-token", blockingResult.getAccessToken());
+
+    // Let the async refresh finish — its older token should be discarded.
+    allowAsyncToFinish.countDown();
+    awaitCondition(
+        "refreshInProgress should be reset after the async refresh completes",
+        () -> {
+          try {
+            Field f = CachedTokenSource.class.getDeclaredField("refreshInProgress");
+            f.setAccessible(true);
+            return !(boolean) f.get(source);
+          } catch (Exception e) {
+            throw new RuntimeException(e);
+          }
+        });
+
+    assertEquals(
+        "newer-blocking-token",
+        source.getToken().getAccessToken(),
+        "The newer blocking token should still be cached after the older async result is discarded");
+  }
+
   /**
    * Builds a CachedTokenSource whose initial token is already stale. The clock is advanced past the
    * dynamic staleAfter threshold so the very first getToken call triggers an async refresh.