Skip to content

fix: preserve shared resources across derived clients #851

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abhinavkr26104 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: preserve shared resources across derived clients #851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ffb0468
docs: fix async pagination examples
abhinavkr26104 Aug 10, 2026
183fd87
fix: preserve shared resources across derived clients
abhinavkr26104 Aug 10, 2026
24dc9ae
fix: keep resources alive for in-flight async requests
abhinavkr26104 Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1343,12 +1343,12 @@ import java.util.concurrent.CompletableFuture;

CompletableFuture<JobListPageAsync> 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);
Expand All @@ -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);
})
Expand Down
5 changes: 4 additions & 1 deletion openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientAsyncImpl.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
5 changes: 4 additions & 1 deletion openai-java-core/src/main/kotlin/com/openai/client/OpenAIClientImpl.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment on lines +217 to +218

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Close initialized async views before releasing options

When async() has been accessed on a sync client, the lazy child constructs its own clientOptionsWithUserAgent and retains the same ClientOptionsResources; this close path only releases the parent UA options and the original options, so the refcount stays above zero and explicit client.close() leaves the HTTP client/executor/sleeper open as long as the parent still holds that lazy child. Close initialized child views as well, or avoid giving them a separately retained options instance.

Useful? React with 👍 / 👎.

}

class WithRawResponseImpl internal constructor(private val clientOptions: ClientOptions) :
OpenAIClient.WithRawResponse {
Expand Down
109 changes: 103 additions & 6 deletions openai-java-core/src/main/kotlin/com/openai/core/ClientOptions.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment on lines +70 to +74

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Release remaining resources after authenticator failures

When a provider HttpRequestAuthenticator.close() throws while closing the last owning options/client, this first release propagates immediately, so the HTTP client, stream executor, and sleeper are never released and later close() calls are no-ops through ClientOptionsCloseAction. The previous AuthenticatingHttpClient.close() path still closed the delegate after authenticator failure, so please keep releasing the remaining resources with suppressed exceptions in this failure path.

Useful? React with 👍 / 👎.

}
}

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(
Expand Down Expand Up @@ -141,12 +191,18 @@ 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()
}
// Async request futures retain the HTTP client chain, not this options object. Observe the
// chain so phantom cleanup cannot close resources while an in-flight request still uses it.
closeWhenPhantomReachable(httpClient, closeAction)

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep non-retryable async requests from being cleaned up

For async uploads with InputStream multipart parts, HttpRequestBody.repeatable() is false and RetryingHttpClient.executeAsync() returns the delegate responseFuture directly instead of a future that captures this RetryingHttpClient; registering the cleaner on this wrapper can therefore still release the shared HTTP client/sleeper while the only remaining reference is the in-flight delegate future. Fresh evidence for re-raising the earlier in-flight concern is this non-retryable branch, which bypasses the new reachability chain entirely.

Useful? React with 👍 / 👎.

}

/**
Expand Down Expand Up @@ -217,6 +273,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 {
Expand All @@ -226,6 +287,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
Comment on lines +290 to +294

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep copied resources alive while a builder is pending

When code keeps only the ClientOptions.Builder returned by toBuilder() and lets the source ClientOptions become unreachable before calling build(), these copied resource handles are not counted as references. The new ClientOptions cleaner can therefore release and mark the shared ClientOptionsResources closed while the builder still intends to reuse them, and the later build() hits retain() on a closed resource instead of producing the derived options. Retain the resources for the builder's lifetime or keep the source options strongly reachable until build() transfers ownership.

Useful? React with 👍 / 👎.

clock = clientOptions.clock
baseUrl = clientOptions.baseUrl
headers = clientOptions.headers.toBuilder()
Expand Down Expand Up @@ -256,6 +322,7 @@ private constructor(
*/
fun httpClient(httpClient: HttpClient) = apply {
this.httpClient = PhantomReachableClosingHttpClient(httpClient)
this.httpClientResource = null
}

/**
Expand All @@ -269,6 +336,7 @@ private constructor(
this.httpRequestAuthenticator =
if (httpRequestAuthenticator == null) null
else PhantomReachableClosingHttpRequestAuthenticator(httpRequestAuthenticator)
this.httpRequestAuthenticatorResource = null
}

/**
Expand Down Expand Up @@ -302,6 +370,7 @@ private constructor(
if (streamHandlerExecutor is ExecutorService)
PhantomReachableExecutorService(streamHandlerExecutor)
else streamHandlerExecutor
this.streamHandlerExecutorResource = null
}

/**
Expand All @@ -313,7 +382,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.
Expand Down Expand Up @@ -393,6 +465,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)`. */
Expand All @@ -406,6 +479,7 @@ private constructor(
fun credential(credential: Credential) = apply {
this.apiKey = null
this.credential = credential
this.workloadIdentityAuthResource = null
}

fun azureServiceVersion(azureServiceVersion: AzureOpenAIServiceVersion) = apply {
Expand Down Expand Up @@ -434,6 +508,7 @@ private constructor(

fun workloadIdentity(workloadIdentity: WorkloadIdentity?) = apply {
this.workloadIdentity = workloadIdentity
this.workloadIdentityAuthResource = null

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Retain inherited workload identity auth when it is still used

When options created with workloadIdentity(...) are copied and the modifier calls workloadIdentity(null) without replacing the credential, from() has already copied the existing WorkloadIdentityCredential, so effectiveCredential() still reuses the same WorkloadIdentityAuth. Clearing workloadIdentityAuthResource here makes the derived options create an independent close resource for that same auth object, so closing the derived options can close the provider while the original options still rely on it, and closing both can close it twice.

Useful? React with 👍 / 👎.

}

/** Alias for calling [Builder.workloadIdentity] with `workloadIdentity.orElse(null)`. */
Expand Down Expand Up @@ -701,6 +776,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
Expand Down Expand Up @@ -756,6 +856,7 @@ private constructor(
organization,
project,
webhookSecret,
resources,
)
}
}
Expand All @@ -770,11 +871,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 {
Expand Down
Loading