SK-3133: Use the generated RetryInterceptor + fix logging/dotenv latency bottlenecks

common/src/main/java/com/skyflow/BaseVaultClient.java

@@ -14,7 +14,6 @@
 import com.skyflow.utils.BaseUtils;
 import com.skyflow.utils.logger.LogUtil;
 import com.skyflow.utils.validations.BaseValidations;
-import io.github.cdimascio.dotenv.Dotenv;
 import io.github.cdimascio.dotenv.DotenvException;
 import okhttp3.ConnectionPool;
 import okhttp3.OkHttpClient;
@@ -63,11 +62,7 @@ protected synchronized void prioritiseCredentials(BaseCredentials vaultSpecificC
             } else if (this.commonCredentials != null) {
                 this.finalCredentials = this.commonCredentials;
             } else {
-                String sysCredentials = System.getenv(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
-                if (sysCredentials == null) {
-                    Dotenv dotenv = Dotenv.load();
-                    sysCredentials = dotenv.get(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
-                }
+                String sysCredentials = BaseUtils.resolveEnvOrDotenv(BaseConstants.ENV_CREDENTIALS_KEY_NAME);
                 if (sysCredentials == null) {
                     throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyCredentials.getMessage());
                 } else {

common/src/main/java/com/skyflow/utils/BaseUtils.java

@@ -21,8 +21,63 @@
 import com.skyflow.logs.InfoLogs;
 import com.skyflow.serviceaccount.util.BearerToken;
 import com.skyflow.utils.logger.LogUtil;
+import io.github.cdimascio.dotenv.Dotenv;
+import io.github.cdimascio.dotenv.DotenvException;
 
 public class BaseUtils {
+
+    // Memoized .env: Dotenv.load() does a filesystem read every time it's called, and several
+    // call sites resolve a setting this way on every single SDK request -- re-reading a file
+    // whose contents never change for the life of the process turned those into repeated,
+    // uncached, blocking disk I/O on the hot path. Loaded at most once per JVM; `dotenvAttempted`
+    // also memoizes the "no .env file present" outcome so a missing file isn't retried either.
+    private static volatile boolean dotenvAttempted = false;
+    private static volatile Dotenv cachedDotenv = null;
+
+    private static Dotenv memoizedDotenv() {
+        if (!dotenvAttempted) {
+            synchronized (BaseUtils.class) {
+                if (!dotenvAttempted) {
+                    try {
+                        cachedDotenv = Dotenv.load();
+                    } catch (DotenvException e) {
+                        cachedDotenv = null; // no .env file in the working directory
+                    }
+                    dotenvAttempted = true;
+                }
+            }
+        }
+        return cachedDotenv;
+    }
+
+    /**
+     * Resolves {@code key} from the process environment first, falling back to the (memoized)
+     * {@code .env} file if present. Returns null if found in neither.
+     */
+    public static String resolveEnvOrDotenv(String key) {
+        String value = System.getenv(key);
+        if (value == null) {
+            Dotenv dotenv = memoizedDotenv();
+            if (dotenv != null) {
+                value = dotenv.get(key);
+            }
+        }
+        return value;
+    }
+
+    /**
+     * Test-only: forces the next {@link #resolveEnvOrDotenv} call to re-read the {@code .env}
+     * file from disk instead of reusing the memoized one. Production code always wants the
+     * memoized behavior (a project's {@code .env} doesn't change while the process is running);
+     * this exists purely so tests that rewrite {@code .env} mid-run can observe the new content
+     * without restarting the JVM.
+     */
+    public static void resetDotenvCacheForTests() {
+        synchronized (BaseUtils.class) {
+            dotenvAttempted = false;
+            cachedDotenv = null;
+        }
+    }
     public static String generateBearerToken(BaseCredentials credentials) throws SkyflowException {
         if (credentials.getPath() != null) {
             BearerToken.BearerTokenBuilder builder = BearerToken.builder()

common/src/main/java/com/skyflow/utils/logger/AsyncConsoleHandler.java

@@ -0,0 +1,82 @@
+package com.skyflow.utils.logger;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.ErrorManager;
+import java.util.logging.Handler;
+import java.util.logging.LogRecord;
+
+/**
+ * Wraps a delegate {@link Handler} (in practice a {@link java.util.logging.ConsoleHandler}) so that
+ * {@link #publish(LogRecord)} never performs blocking I/O on the calling thread.
+ * <p>
+ * {@code ConsoleHandler.publish} writes to and flushes the underlying stream synchronously, and does
+ * so under a lock shared by every thread using the logger. Since {@code LogUtil.printInfoLog}/etc. are
+ * called on every SDK request (often several times per call), that turns console logging into a
+ * per-request blocking-I/O + lock-contention point under concurrent load. This handler hands each
+ * {@link LogRecord} off to a single background daemon thread instead, which performs the actual write;
+ * the calling thread only enqueues.
+ * <p>
+ * The handoff never blocks or applies backpressure to the caller: if the queue is momentarily full
+ * (a sustained logging flood, or the writer thread stalled) the record is dropped rather than slowing
+ * down request-serving threads — logging must never become the bottleneck it was flagged for.
+ */
+final class AsyncConsoleHandler extends Handler {
+
+    /** Bounds worst-case memory use if the writer thread falls behind; excess records are dropped. */
+    private static final int QUEUE_CAPACITY = 10_000;
+
+    private final Handler delegate;
+    private final LinkedBlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
+    private final Thread writer;
+    private volatile boolean closed = false;
+
+    AsyncConsoleHandler(Handler delegate) {
+        this.delegate = delegate;
+        setLevel(delegate.getLevel());
+        this.writer = new Thread(this::drain, "skyflow-sdk-log-writer");
+        this.writer.setDaemon(true);
+        this.writer.start();
+    }
+
+    @Override
+    public void publish(LogRecord record) {
+        if (closed || !isLoggable(record)) {
+            return;
+        }
+        // offer() never blocks: a full queue means "drop", never "wait".
+        queue.offer(record);
+    }
+
+    private void drain() {
+        try {
+            while (true) {
+                LogRecord record = queue.take();
+                try {
+                    delegate.publish(record);
+                } catch (RuntimeException e) {
+                    reportError(null, e, ErrorManager.WRITE_FAILURE);
+                }
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    @Override
+    public void flush() {
+        // Best-effort: drain what's queued right now onto the delegate, then flush it.
+        LogRecord record;
+        while ((record = queue.poll()) != null) {
+            delegate.publish(record);
+        }
+        delegate.flush();
+    }
+
+    @Override
+    public void close() {
+        closed = true;
+        writer.interrupt();
+        flush();
+        delegate.close();
+    }
+}

common/src/main/java/com/skyflow/utils/logger/LogUtil.java

@@ -35,7 +35,10 @@ public synchronized String format(LogRecord logRecord) {
         consoleHandler.setFormatter(formatter);
         consoleHandler.setLevel(Level.CONFIG);
 
-        LOGGER.addHandler(consoleHandler);
+        // The actual write+flush to the console happens on a background thread, so per-call
+        // logging never blocks (or lock-contends on) the request-serving thread. See
+        // AsyncConsoleHandler's class doc for why this matters under concurrent load.
+        LOGGER.addHandler(new AsyncConsoleHandler(consoleHandler));
         LOGGER.setLevel(logLevelToLoggerLevelMap(logLevel));
         printInfoLog(InfoLogs.LOGGER_SETUP_DONE.getLog());
     }

common/src/test/java/com/skyflow/BaseVaultClientTests.java

@@ -6,6 +6,7 @@
 import com.skyflow.errors.SkyflowException;
 import com.skyflow.logs.ErrorLogs;
 import com.skyflow.utils.BaseConstants;
+import com.skyflow.utils.BaseUtils;
 import okhttp3.Call;
 import okhttp3.Connection;
 import okhttp3.Interceptor;
@@ -38,6 +39,7 @@ public class BaseVaultClientTests {
     public void saveEnvFileState() throws IOException {
         File f = new File(ENV_FILE);
         originalEnvContent = f.exists() ? Files.readAllBytes(Paths.get(ENV_FILE)) : null;
+        BaseUtils.resetDotenvCacheForTests(); // see its javadoc: .env is otherwise memoized JVM-wide
     }
 
     @After

flowvault/src/main/java/com/skyflow/VaultClient.java

@@ -5,11 +5,12 @@
 import com.skyflow.errors.SkyflowException;
 import com.skyflow.generated.rest.ApiClient;
 import com.skyflow.generated.rest.ApiClientBuilder;
+import com.skyflow.generated.rest.core.RetryInterceptor;
 import com.skyflow.generated.rest.resources.flowservice.FlowserviceClient;
 import com.skyflow.generated.rest.resources.records.RecordsClient;
-import com.skyflow.utils.SkyflowRetryInterceptor;
 import com.skyflow.utils.Utils;
 
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 import okhttp3.ConnectionPool;
@@ -33,6 +34,11 @@ public class VaultClient extends BaseVaultClient<VaultConfig> {
     private static final int DEFAULT_MAX_RETRIES = 0;
     private static final long DEFAULT_INITIAL_RETRY_DELAY_MILLIS = 500L;
     private static final long DEFAULT_MAX_RETRY_DELAY_MILLIS = 2000L;
+    // Not yet exposed as a VaultConfig/builder setting, so hardcoded here rather than left as
+    // Optional.empty() - passing it explicitly keeps the choice visible in our own code instead of
+    // depending on RetryInterceptor's internal default, which is free to change on a future
+    // regeneration since it is generated code we do not maintain.
+    private static final double RETRY_JITTER_FACTOR = 0.2;
 
     protected VaultClient(VaultConfig vaultConfig, Credentials credentials) throws SkyflowException {
         super(vaultConfig, credentials);
@@ -159,18 +165,23 @@ protected void updateExecutorInHTTP() throws SkyflowException {
             Integer writeTimeout = resolveNullableInt(vaultConfig.getWriteTimeout(), commonWriteTimeout);
 
             // Negative timeout/retry values reach here straight from public config setters with
-            // no validation of their own; our own SkyflowRetryInterceptor throws IllegalArgumentException
-            // and OkHttp's own Builder throws IllegalStateException for those — translate both (and
-            // anything else unexpected from this construction) to SkyflowException so every failure
-            // mode from this SDK is a SkyflowException, never a raw one.
+            // no validation of their own; the generated RetryInterceptor validates initial/max delay
+            // but not maxRetries itself (a negative value would just behave as zero retries), and
+            // OkHttp's own Builder throws IllegalStateException for negative timeouts — translate
+            // all of these (and anything else unexpected from this construction) to SkyflowException
+            // so every failure mode from this SDK is a SkyflowException, never a raw one.
             try {
+                if (maxRetries < 0) {
+                    throw new IllegalArgumentException("maxRetries must be non-negative");
+                }
                 OkHttpClient.Builder httpBuilder = new OkHttpClient.Builder()
                         .connectionPool(new ConnectionPool(10, 1, TimeUnit.MINUTES))
                         // Overall ceiling; bounds the whole call including retries.
                         .callTimeout(timeoutSeconds, TimeUnit.SECONDS)
                         // OUTER: retries. Must wrap the auth interceptor so each attempt re-reads the
                         // (possibly refreshed) bearer token rather than replaying a stale one.
-                        .addInterceptor(new SkyflowRetryInterceptor(maxRetries, initialRetryDelayMillis, maxRetryDelayMillis))
+                        .addInterceptor(new RetryInterceptor(maxRetries, Optional.of(initialRetryDelayMillis),
+                                Optional.of(maxRetryDelayMillis), Optional.of(RETRY_JITTER_FACTOR)))
                         .addInterceptor(chain -> {  // INNER: auth
                             Request requestWithAuth = chain.request().newBuilder()
                                     .header("Authorization", "Bearer " + this.token)