BackgroundIdentity: do not poison the auth state when a token wait is interrupted

core/src/main/java/tech/ydb/core/auth/BackgroundIdentity.java

@@ -2,6 +2,7 @@
 
 import java.time.Clock;
 import java.time.Instant;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -72,6 +73,7 @@ public void close() {
     }
 
     private State updateState(State current, State next) {
+        Objects.requireNonNull(next, "next state cannot be null");
         if (state.compareAndSet(current, next)) {
             next.init();
         }
@@ -92,7 +94,8 @@ private <T> T unwrap(CompletableFuture<T> future) {
         } catch (InterruptedException ex) {
             logger.error("updating of authentication token was interrupted", ex);
             Thread.currentThread().interrupt();
-            return null;
+            // returning null here would poison the state reference and break every following getToken()
+            throw new RuntimeException("authentication update was interrupted", ex);
         }
     }
 

core/src/test/java/tech/ydb/core/auth/BackgroundIdentityTest.java

@@ -0,0 +1,68 @@
+package tech.ydb.core.auth;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class BackgroundIdentityTest {
+    private final Instant now = Instant.EPOCH;
+    private final Clock clock = Clock.fixed(now, ZoneId.of("UTC"));
+
+    private static class MockedRpc implements BackgroundIdentity.Rpc {
+        private final CompletableFuture<Token> tokenFuture = new CompletableFuture<>();
+
+        @Override
+        public CompletableFuture<Token> getTokenAsync() {
+            return tokenFuture;
+        }
+
+        @Override
+        public int getTimeoutSeconds() {
+            return 60;
+        }
+    }
+
+    @Test(timeout = 30_000)
+    public void interruptedGetTokenDoesNotBreakIdentity() throws InterruptedException {
+        MockedRpc rpc = new MockedRpc();
+        BackgroundIdentity identity = new BackgroundIdentity(clock, rpc);
+
+        AtomicReference<Throwable> caught = new AtomicReference<>();
+        Thread reader = new Thread(() -> {
+            try {
+                identity.getToken();
+            } catch (Throwable th) {
+                caught.set(th);
+            }
+        });
+
+        // the login never answers, so the reader blocks in the sync state and gets interrupted there
+        reader.start();
+        // the reader may not have reached the await yet, interrupting early is handled the same way
+        reader.interrupt();
+        reader.join();
+
+        Assert.assertNotNull("interrupted getToken must report a failure", caught.get());
+        Assert.assertFalse(
+                "interrupt must not surface as a NullPointerException",
+                caught.get() instanceof NullPointerException
+        );
+
+        // the identity must recover once the login completes
+        rpc.getTokenAsync().complete(
+                new BackgroundIdentity.Rpc.Token(
+                        "token-value",
+                        now.plus(Duration.ofHours(2)),
+                        now.plus(Duration.ofHours(1))
+                )
+        );
+
+        Assert.assertEquals("token-value", identity.getToken());
+    }
+}