From ffb0468720a3e9ed7eae725b59b5f2cce595f85e Mon Sep 17 00:00:00 2001 From: Abhinav Kumar Singh Date: Mon, 10 Aug 2026 11:29:44 +0530 Subject: [PATCH 1/3] docs: fix async pagination examples --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 99c129546..084778561 100644 --- a/README.md +++ b/README.md @@ -1343,12 +1343,12 @@ import java.util.concurrent.CompletableFuture; CompletableFuture pageFuture = client.async().fineTuning().jobs().list(); -pageFuture.thenRun(page -> page.autoPager().subscribe(job -> { +pageFuture.thenAccept(page -> page.autoPager().subscribe(job -> { System.out.println(job); })); // If you need to handle errors or completion of the stream -pageFuture.thenRun(page -> page.autoPager().subscribe(new AsyncStreamResponse.Handler<>() { +pageFuture.thenAccept(page -> page.autoPager().subscribe(new AsyncStreamResponse.Handler<>() { @Override public void onNext(FineTuningJob job) { System.out.println(job); @@ -1366,7 +1366,7 @@ pageFuture.thenRun(page -> page.autoPager().subscribe(new AsyncStreamResponse.Ha })); // Or use futures -pageFuture.thenRun(page -> page.autoPager() +pageFuture.thenAccept(page -> page.autoPager() .subscribe(job -> { System.out.println(job); }) From 183fd87ec8d61079d1162635cb2e91c6c68dbc4e Mon Sep 17 00:00:00 2001 From: Abhinav Kumar Singh Date: Mon, 10 Aug 2026 12:06:28 +0530 Subject: [PATCH 2/3] fix: preserve shared resources across derived clients --- .../openai/client/OpenAIClientAsyncImpl.kt | 5 +- .../com/openai/client/OpenAIClientImpl.kt | 5 +- .../kotlin/com/openai/core/ClientOptions.kt | 107 +++++++++++++++++- .../com/openai/core/ClientOptionsTest.kt | 91 +++++++++++++++ 4 files changed, 200 insertions(+), 8 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientAsyncImpl.kt b/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientAsyncImpl.kt index 78c78f6aa..a5c8e76c4 100644 --- a/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientAsyncImpl.kt +++ b/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientAsyncImpl.kt @@ -234,7 +234,10 @@ class OpenAIClientAsyncImpl(private val clientOptions: ClientOptions) : OpenAICl override fun videos(): VideoServiceAsync = videos - override fun close() = clientOptions.close() + override fun close() { + clientOptionsWithUserAgent.close() + if (clientOptionsWithUserAgent !== clientOptions) clientOptions.close() + } class WithRawResponseImpl internal constructor(private val clientOptions: ClientOptions) : OpenAIClientAsync.WithRawResponse { diff --git a/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientImpl.kt b/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientImpl.kt index 52f28f315..a2929ec3b 100644 --- a/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientImpl.kt +++ b/openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientImpl.kt @@ -213,7 +213,10 @@ class OpenAIClientImpl(private val clientOptions: ClientOptions) : OpenAIClient override fun videos(): VideoService = videos - override fun close() = clientOptions.close() + override fun close() { + clientOptionsWithUserAgent.close() + if (clientOptionsWithUserAgent !== clientOptions) clientOptions.close() + } class WithRawResponseImpl internal constructor(private val clientOptions: ClientOptions) : OpenAIClient.WithRawResponse { diff --git a/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt b/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt index fe00a3363..179282df6 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt @@ -30,9 +30,59 @@ import java.util.concurrent.Executor import java.util.concurrent.ExecutorService import java.util.concurrent.Executors import java.util.concurrent.ThreadFactory +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong import kotlin.jvm.optionals.getOrNull +private class ClientOptionsResource(private val close: () -> Unit) { + private var references = 1 + private var closed = false + + @Synchronized + fun retain() { + check(!closed) { "Cannot retain a closed client resource" } + references++ + } + + fun release() { + val shouldClose = + synchronized(this) { + check(references > 0) { "Client resource released too many times" } + references-- + if (references == 0) { + closed = true + true + } else false + } + + if (shouldClose) close() + } +} + +private class ClientOptionsResources( + val httpClient: ClientOptionsResource, + val httpRequestAuthenticator: ClientOptionsResource?, + val workloadIdentityAuth: ClientOptionsResource?, + val streamHandlerExecutor: ClientOptionsResource, + val sleeper: ClientOptionsResource, +) { + fun release() { + httpRequestAuthenticator?.release() + workloadIdentityAuth?.release() + httpClient.release() + streamHandlerExecutor.release() + sleeper.release() + } +} + +private class ClientOptionsCloseAction(private val resources: ClientOptionsResources) : () -> Unit { + private val closed = AtomicBoolean(false) + + override fun invoke() { + if (closed.compareAndSet(false, true)) resources.release() + } +} + /** A class representing the SDK client configuration. */ class ClientOptions private constructor( @@ -141,12 +191,16 @@ private constructor( private val organization: String?, private val project: String?, private val webhookSecret: String?, + private val resources: ClientOptionsResources, ) { + private val closeAction = ClientOptionsCloseAction(resources) + init { if (checkJacksonVersionCompatibility) { checkJacksonVersionCompatibility() } + closeWhenPhantomReachable(this, closeAction) } /** @@ -217,6 +271,11 @@ private constructor( private var project: String? = null private var webhookSecret: String? = null private var workloadIdentity: WorkloadIdentity? = null + private var httpClientResource: ClientOptionsResource? = null + private var httpRequestAuthenticatorResource: ClientOptionsResource? = null + private var workloadIdentityAuthResource: ClientOptionsResource? = null + private var streamHandlerExecutorResource: ClientOptionsResource? = null + private var sleeperResource: ClientOptionsResource? = null @JvmSynthetic internal fun from(clientOptions: ClientOptions) = apply { @@ -226,6 +285,11 @@ private constructor( jsonMapper = clientOptions.jsonMapper streamHandlerExecutor = clientOptions.streamHandlerExecutor sleeper = clientOptions.sleeper + httpClientResource = clientOptions.resources.httpClient + httpRequestAuthenticatorResource = clientOptions.resources.httpRequestAuthenticator + workloadIdentityAuthResource = clientOptions.resources.workloadIdentityAuth + streamHandlerExecutorResource = clientOptions.resources.streamHandlerExecutor + sleeperResource = clientOptions.resources.sleeper clock = clientOptions.clock baseUrl = clientOptions.baseUrl headers = clientOptions.headers.toBuilder() @@ -256,6 +320,7 @@ private constructor( */ fun httpClient(httpClient: HttpClient) = apply { this.httpClient = PhantomReachableClosingHttpClient(httpClient) + this.httpClientResource = null } /** @@ -269,6 +334,7 @@ private constructor( this.httpRequestAuthenticator = if (httpRequestAuthenticator == null) null else PhantomReachableClosingHttpRequestAuthenticator(httpRequestAuthenticator) + this.httpRequestAuthenticatorResource = null } /** @@ -302,6 +368,7 @@ private constructor( if (streamHandlerExecutor is ExecutorService) PhantomReachableExecutorService(streamHandlerExecutor) else streamHandlerExecutor + this.streamHandlerExecutorResource = null } /** @@ -313,7 +380,10 @@ private constructor( * * This class takes ownership of the sleeper and closes it when closed. */ - fun sleeper(sleeper: Sleeper) = apply { this.sleeper = PhantomReachableSleeper(sleeper) } + fun sleeper(sleeper: Sleeper) = apply { + this.sleeper = PhantomReachableSleeper(sleeper) + this.sleeperResource = null + } /** * The clock to use for operations that require timing, like retries. @@ -393,6 +463,7 @@ private constructor( fun apiKey(apiKey: String?) = apply { this.apiKey = apiKey this.credential = apiKey?.let { BearerTokenCredential.create(it) } + this.workloadIdentityAuthResource = null } /** Alias for calling [Builder.apiKey] with `apiKey.orElse(null)`. */ @@ -406,6 +477,7 @@ private constructor( fun credential(credential: Credential) = apply { this.apiKey = null this.credential = credential + this.workloadIdentityAuthResource = null } fun azureServiceVersion(azureServiceVersion: AzureOpenAIServiceVersion) = apply { @@ -434,6 +506,7 @@ private constructor( fun workloadIdentity(workloadIdentity: WorkloadIdentity?) = apply { this.workloadIdentity = workloadIdentity + this.workloadIdentityAuthResource = null } /** Alias for calling [Builder.workloadIdentity] with `workloadIdentity.orElse(null)`. */ @@ -701,6 +774,31 @@ private constructor( val effectiveWorkloadIdentityAuth = (credential as? WorkloadIdentityCredential)?.getAuth() + val resources = + ClientOptionsResources( + httpClient = + httpClientResource?.also { it.retain() } + ?: ClientOptionsResource { httpClient.close() }, + httpRequestAuthenticator = + httpRequestAuthenticatorResource?.also { it.retain() } + ?: httpRequestAuthenticator?.let { + ClientOptionsResource { it.close() } + }, + workloadIdentityAuth = + workloadIdentityAuthResource?.also { it.retain() } + ?: effectiveWorkloadIdentityAuth?.let { + ClientOptionsResource { it.close() } + }, + streamHandlerExecutor = + streamHandlerExecutorResource?.also { it.retain() } + ?: ClientOptionsResource { + (streamHandlerExecutor as? ExecutorService)?.shutdown() + }, + sleeper = + sleeperResource?.also { it.retain() } + ?: ClientOptionsResource { sleeper.close() }, + ) + val loggingDelegate = if (httpRequestAuthenticator != null) httpClient else @@ -756,6 +854,7 @@ private constructor( organization, project, webhookSecret, + resources, ) } } @@ -770,11 +869,7 @@ private constructor( * releases threads and connections if they remain idle, but if you are writing an application * that needs to aggressively release unused resources, then you may call this method. */ - fun close() { - httpClient.close() - (streamHandlerExecutor as? ExecutorService)?.shutdown() - sleeper.close() - } + fun close() = closeAction() @JvmSynthetic internal fun securityHeaders(security: SecurityOptions): Headers { diff --git a/openai-java-core/src/test/kotlin/com/openai/core/ClientOptionsTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/ClientOptionsTest.kt index ace2f857f..2fe9d9728 100644 --- a/openai-java-core/src/test/kotlin/com/openai/core/ClientOptionsTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/core/ClientOptionsTest.kt @@ -7,12 +7,15 @@ import com.openai.auth.SubjectTokenProvider import com.openai.auth.SubjectTokenType import com.openai.auth.WorkloadIdentity import com.openai.azure.credential.AzureApiKeyCredential +import com.openai.client.OpenAIClientAsyncImpl +import com.openai.client.OpenAIClientImpl import com.openai.core.http.HttpClient import com.openai.core.http.HttpRequest import com.openai.core.http.HttpRequestAuthenticator import com.openai.credential.BearerTokenCredential import com.openai.credential.WorkloadIdentityCredential import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutorService import org.assertj.core.api.Assertions.assertThat import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows @@ -20,6 +23,7 @@ import org.junit.jupiter.api.extension.ExtendWith import org.mockito.junit.jupiter.MockitoExtension import org.mockito.kotlin.mock import org.mockito.kotlin.never +import org.mockito.kotlin.times import org.mockito.kotlin.verify @ExtendWith(MockitoExtension::class) @@ -210,6 +214,93 @@ internal class ClientOptionsTest { .containsExactly("another My Organization") } + @Test + fun toBuilder_closingDerivedOptionsDoesNotCloseSharedResources() { + val executor = mock() + val sleeper = mock() + val original = + ClientOptions.builder() + .httpClient(httpClient) + .streamHandlerExecutor(executor) + .sleeper(sleeper) + .apiKey("My API Key") + .build() + val derived = original.toBuilder().baseUrl("https://example.test").build() + + derived.close() + + verify(httpClient, never()).close() + verify(executor, never()).shutdown() + verify(sleeper, never()).close() + + original.close() + + verify(httpClient).close() + verify(executor).shutdown() + verify(sleeper).close() + } + + @Test + fun toBuilder_closingOriginalOptionsDoesNotCloseResourcesUsedByDerivedOptions() { + val executor = mock() + val sleeper = mock() + val original = + ClientOptions.builder() + .httpClient(httpClient) + .streamHandlerExecutor(executor) + .sleeper(sleeper) + .apiKey("My API Key") + .build() + val derived = original.toBuilder().build() + + original.close() + + verify(httpClient, never()).close() + verify(executor, never()).shutdown() + verify(sleeper, never()).close() + + derived.close() + derived.close() + + verify(httpClient, times(1)).close() + verify(executor, times(1)).shutdown() + verify(sleeper, times(1)).close() + } + + @Test + fun toBuilder_replacingHttpClientOnlyClosesReplacement() { + val replacement = mock() + val original = ClientOptions.builder().httpClient(httpClient).apiKey("My API Key").build() + val derived = original.toBuilder().httpClient(replacement).build() + + derived.close() + + verify(replacement).close() + verify(httpClient, never()).close() + + original.close() + + verify(httpClient).close() + } + + @Test + fun syncClientCloseClosesInternalUserAgentOptions() { + val options = ClientOptions.builder().httpClient(httpClient).apiKey("My API Key").build() + + OpenAIClientImpl(options).close() + + verify(httpClient).close() + } + + @Test + fun asyncClientCloseClosesInternalUserAgentOptions() { + val options = ClientOptions.builder().httpClient(httpClient).apiKey("My API Key").build() + + OpenAIClientAsyncImpl(options).close() + + verify(httpClient).close() + } + @Test fun toBuilder_whenOriginalClientOptionsGarbageCollected_doesNotCloseOriginalClient() { var clientOptions = From a80385468cd87f9b41326c914f450c8f4a4d5cf1 Mon Sep 17 00:00:00 2001 From: Abhinav Kumar Singh Date: Mon, 10 Aug 2026 12:23:36 +0530 Subject: [PATCH 3/3] fix: clear workload identity refresh state on sync failures --- .../com/openai/auth/WorkloadIdentityAuth.kt | 12 +- .../openai/auth/WorkloadIdentityAuthTest.kt | 131 ++++++++++++++++++ 2 files changed, 139 insertions(+), 4 deletions(-) diff --git a/openai-java-core/src/main/kotlin/com/openai/auth/WorkloadIdentityAuth.kt b/openai-java-core/src/main/kotlin/com/openai/auth/WorkloadIdentityAuth.kt index 51d97e950..ee2c36ce8 100644 --- a/openai-java-core/src/main/kotlin/com/openai/auth/WorkloadIdentityAuth.kt +++ b/openai-java-core/src/main/kotlin/com/openai/auth/WorkloadIdentityAuth.kt @@ -236,11 +236,15 @@ internal class WorkloadIdentityAuth( } private fun refreshTokenAsync(): CompletableFuture { - return config.provider.getTokenAsync(httpClient, jsonMapper).thenCompose { subjectToken -> - val request = buildTokenExchangeRequest(subjectToken) - httpClient.executeAsync(request).thenApply { response -> - response.use { processTokenExchangeResponse(it) } + return try { + config.provider.getTokenAsync(httpClient, jsonMapper).thenCompose { subjectToken -> + val request = buildTokenExchangeRequest(subjectToken) + httpClient.executeAsync(request).thenApply { response -> + response.use { processTokenExchangeResponse(it) } + } } + } catch (e: Exception) { + CompletableFuture().also { it.completeExceptionally(e) } } } diff --git a/openai-java-core/src/test/kotlin/com/openai/auth/WorkloadIdentityAuthTest.kt b/openai-java-core/src/test/kotlin/com/openai/auth/WorkloadIdentityAuthTest.kt index 075239dd4..4eeb64dde 100644 --- a/openai-java-core/src/test/kotlin/com/openai/auth/WorkloadIdentityAuthTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/auth/WorkloadIdentityAuthTest.kt @@ -9,7 +9,9 @@ import com.openai.errors.BadRequestException import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream import java.util.concurrent.CompletableFuture +import java.util.concurrent.atomic.AtomicInteger import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows import org.junit.jupiter.api.extension.ExtendWith @@ -67,6 +69,135 @@ internal class WorkloadIdentityAuthTest { verifyNoInteractions(httpClient) } + @Test + fun getTokenAsync_clearsRefreshStateAfterProviderThrowsSynchronously() { + val failure = IllegalStateException("provider failed") + val subjectToken = "subject-token" + val accessToken = "test-access-token" + val providerCalls = AtomicInteger() + val provider = + object : SubjectTokenProvider { + override fun tokenType() = SubjectTokenType.JWT + + override fun getToken(httpClient: HttpClient, jsonMapper: JsonMapper): String = + subjectToken + + override fun getTokenAsync( + httpClient: HttpClient, + jsonMapper: JsonMapper, + ): CompletableFuture { + if (providerCalls.getAndIncrement() == 0) throw failure + return CompletableFuture.completedFuture(subjectToken) + } + } + + val response = + mockResponse( + 200, + """ + { + "access_token": "$accessToken", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "token_type": "Bearer", + "expires_in": 3600 + } + """.trimIndent(), + ) + whenever(httpClient.executeAsync(any())) + .thenReturn(CompletableFuture.completedFuture(response)) + + val auth = + WorkloadIdentityAuth( + config = + WorkloadIdentity.builder() + .clientId("client-id") + .identityProviderId("provider-id") + .serviceAccountId("service-account-id") + .provider(provider) + .build(), + httpClient = httpClient, + jsonMapper = JsonMapper(), + ) + + val first = auth.getTokenAsync() + assertThatThrownBy { first.join() }.hasCauseSameAs(failure) + + assertThat(auth.getTokenAsync().join()).isEqualTo(accessToken) + assertThat(providerCalls.get()).isEqualTo(2) + } + + @Test + fun getTokenAsync_clearsBackgroundRefreshStateAfterProviderThrowsSynchronously() { + val failure = IllegalStateException("provider failed") + val subjectToken = "subject-token" + val initialAccessToken = "initial-access-token" + val refreshedAccessToken = "refreshed-access-token" + val providerCalls = AtomicInteger() + val provider = + object : SubjectTokenProvider { + override fun tokenType() = SubjectTokenType.JWT + + override fun getToken(httpClient: HttpClient, jsonMapper: JsonMapper): String = + subjectToken + + override fun getTokenAsync( + httpClient: HttpClient, + jsonMapper: JsonMapper, + ): CompletableFuture { + if (providerCalls.getAndIncrement() == 0) throw failure + return CompletableFuture.completedFuture(subjectToken) + } + } + + val initialResponse = + mockResponse( + 200, + """ + { + "access_token": "$initialAccessToken", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "token_type": "Bearer", + "expires_in": 1 + } + """.trimIndent(), + ) + val refreshedResponse = + mockResponse( + 200, + """ + { + "access_token": "$refreshedAccessToken", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "token_type": "Bearer", + "expires_in": 3600 + } + """.trimIndent(), + ) + whenever(httpClient.execute(any())).thenReturn(initialResponse) + whenever(httpClient.executeAsync(any())) + .thenReturn(CompletableFuture.completedFuture(refreshedResponse)) + + val auth = + WorkloadIdentityAuth( + config = + WorkloadIdentity.builder() + .clientId("client-id") + .identityProviderId("provider-id") + .serviceAccountId("service-account-id") + .provider(provider) + .refreshBufferSeconds(2) + .build(), + httpClient = httpClient, + jsonMapper = JsonMapper(), + ) + + assertThat(auth.getToken()).isEqualTo(initialAccessToken) + assertThat(auth.getTokenAsync().join()).isEqualTo(initialAccessToken) + assertThat(auth.getTokenAsync().join()).isEqualTo(initialAccessToken) + assertThat(auth.getTokenAsync().join()).isEqualTo(refreshedAccessToken) + assertThat(providerCalls.get()).isEqualTo(2) + } + @Test fun getToken_success() { val subjectToken = "subject-token"