Skip to content

DGS-23437: Enable client omit Confluent-Identity-Pool-Id #2182

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
Tobi Ogunbi (tobiogunbi) wants to merge 12 commits into
base: master
Choose a base branch
from
Open

DGS-23437: Enable client omit Confluent-Identity-Pool-Id #2182

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
1728c15
DGS-23437: Enable client omit Confluent
tobiogunbi Feb 2, 2026
b0940f3
Fix to sync version
tobiogunbi Feb 2, 2026
efd2db4
Fix tests
tobiogunbi Feb 2, 2026
da6ece1
Unasync run
tobiogunbi Feb 2, 2026
2f45d49
restoring
tobiogunbi Feb 3, 2026
001f1fa
unasync
tobiogunbi Feb 3, 2026
1e86713
undo unrelated unasync build
tobiogunbi Feb 3, 2026
2f5b75a
unasync run again
tobiogunbi Feb 3, 2026
ad173f9
unasync
tobiogunbi Feb 3, 2026
a8d4194
make style-fix and helper method
tobiogunbi Feb 3, 2026
7257b6e
test fix
tobiogunbi Feb 3, 2026
98ff9c6
addressing nits and comments
tobiogunbi Feb 18, 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
27 changes: 27 additions & 0 deletions examples/oauth_schema_registry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,33 @@ def custom_oauth_function(config):
custom_sr_client = SchemaRegistryClient(custom_sr_config)
print(custom_sr_client.get_subjects())

# Example: Using union-of-pools with comma-separated pool IDs
Copy link
Copy Markdown
Member

@fangnx Naxin Fang (fangnx) Feb 18, 2026

Choose a reason for hiding this comment

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

nit: I think this should be extracted into a separate function in example. Combining this with the existing one might be confusing, since it's currently building two separate SR clients + calling get_subjects() twice in one call

union_of_pools_config = {
'url': 'https://psrc-123456.us-east-1.aws.confluent.cloud',
'bearer.auth.credentials.source': 'STATIC_TOKEN',
'bearer.auth.token': 'multi-pool-token',
'bearer.auth.logical.cluster': 'lsrc-12345',
'bearer.auth.identity.pool.id': 'pool-abc,pool-def,pool-ghi',
}

union_sr_client = SchemaRegistryClient(union_of_pools_config)
print(union_sr_client.get_subjects())

# Example: Omitting identity pool for auto pool mapping
auto_pool_config = {
'url': 'https://psrc-123456.us-east-1.aws.confluent.cloud',
'bearer.auth.credentials.source': 'OAUTHBEARER',
'bearer.auth.client.id': 'client-id',
'bearer.auth.client.secret': 'client-secret',
'bearer.auth.scope': 'schema_registry',
'bearer.auth.issuer.endpoint.url': 'https://yourauthprovider.com/v1/token',
'bearer.auth.logical.cluster': 'lsrc-12345',
# bearer.auth.identity.pool.id is omitted - SR will use auto pool mapping
}

auto_pool_sr_client = SchemaRegistryClient(auto_pool_config)
print(auto_pool_sr_client.get_subjects())


if __name__ == '__main__':
main()
27 changes: 15 additions & 12 deletions src/confluent_kafka/schema_registry/_async/schema_registry_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,10 +97,10 @@ def __init__(
scope: str,
token_endpoint: str,
logical_cluster: str,
identity_pool: str,
max_retries: int,
retries_wait_ms: int,
retries_max_wait_ms: int,
identity_pool: Optional[str] = None,
):
self.token = None
self.logical_cluster = logical_cluster
Expand All @@ -113,11 +113,13 @@ def __init__(
self.token_expiry_threshold = 0.8

async def get_bearer_fields(self) -> dict:
return {
fields = {
'bearer.auth.token': await self.get_access_token(),
'bearer.auth.logical.cluster': self.logical_cluster,
'bearer.auth.identity.pool.id': self.identity_pool,
}
if self.identity_pool is not None:
fields['bearer.auth.identity.pool.id'] = self.identity_pool
return fields

def token_expired(self) -> bool:
if self.token is None:
Expand Down Expand Up @@ -283,19 +285,17 @@ def __init__(self, conf: dict):
self.auth = None

if self.bearer_auth_credentials_source in {'OAUTHBEARER', 'STATIC_TOKEN'}:
headers = ['bearer.auth.logical.cluster', 'bearer.auth.identity.pool.id']
missing_headers = [header for header in headers if header not in conf_copy]
if missing_headers:
if 'bearer.auth.logical.cluster' not in conf_copy:
raise ValueError(
"Missing required bearer configuration properties: {}".format(", ".join(missing_headers))
"Missing required bearer configuration property: bearer.auth.logical.cluster"
)

logical_cluster = conf_copy.pop('bearer.auth.logical.cluster')
if not isinstance(logical_cluster, str):
raise TypeError("logical cluster must be a str, not " + str(type(logical_cluster)))

Copy link

Copilot AI Feb 2, 2026

Choose a reason for hiding this comment

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

The comment on lines 297-299 describes identity pool as supporting comma-separated values for union-of-pools, but the type check only validates it as a string. Consider adding a comment explaining that comma-separated values are passed as a single string to clarify the expected format.

Suggested change
# Note: identity_pool is always provided and validated as a single
# string. For union-of-pools use cases, multiple identity pool
# IDs should be encoded as a comma-separated list within this
# string. No additional parsing or validation of the individual
# comma-separated values is performed here.

Copilot uses AI. Check for mistakes.
identity_pool = conf_copy.pop('bearer.auth.identity.pool.id')
if not isinstance(identity_pool, str):
identity_pool = conf_copy.pop('bearer.auth.identity.pool.id', None)
if identity_pool is not None and not isinstance(identity_pool, str):
raise TypeError("identity pool id must be a str, not " + str(type(identity_pool)))

if self.bearer_auth_credentials_source == 'OAUTHBEARER':
Expand Down Expand Up @@ -335,10 +335,10 @@ def __init__(self, conf: dict):
self.scope,
self.token_endpoint,
logical_cluster,
identity_pool,
self.max_retries,
self.retries_wait_ms,
self.retries_max_wait_ms,
identity_pool,
)
else: # STATIC_TOKEN
if 'bearer.auth.token' not in conf_copy:
Expand Down Expand Up @@ -412,7 +412,8 @@ async def handle_bearer_auth(self, headers: dict) -> None:
if self.bearer_field_provider is None:
raise ValueError("Bearer field provider is not set")
bearer_fields = await self.bearer_field_provider.get_bearer_fields()
required_fields = ['bearer.auth.token', 'bearer.auth.identity.pool.id', 'bearer.auth.logical.cluster']
# Note: bearer.auth.identity.pool.id is optional; only token and logical.cluster are required
Copy link
Copy Markdown
Member

@fangnx Naxin Fang (fangnx) Feb 18, 2026

Choose a reason for hiding this comment

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

Should logical.cluster be optional as well, since it's only required for CC not but CP? I know we discussed this for the OAuth CPP implementation (https://github.com/confluentinc/libschemaregistry/pull/13/changes) and decided and it should be optional (cc Justin Wang (@Claimundefine)). Wondering if we should fix it in python client as well

Copy link
Copy Markdown
Member Author

@tobiogunbi Tobi Ogunbi (tobiogunbi) Feb 18, 2026

Choose a reason for hiding this comment

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

logical cluster is always required for SDS, maybe I can just include a comment instructing users?

required_fields = ['bearer.auth.token', 'bearer.auth.logical.cluster']

missing_fields = []
for field in required_fields:
Expand All @@ -427,9 +428,11 @@ async def handle_bearer_auth(self, headers: dict) -> None:
)

headers["Authorization"] = "Bearer {}".format(bearer_fields['bearer.auth.token'])
headers['Confluent-Identity-Pool-Id'] = bearer_fields['bearer.auth.identity.pool.id']
headers['target-sr-cluster'] = bearer_fields['bearer.auth.logical.cluster']

if 'bearer.auth.identity.pool.id' in bearer_fields:
headers['Confluent-Identity-Pool-Id'] = bearer_fields['bearer.auth.identity.pool.id']

async def get(self, url: str, query: Optional[dict] = None) -> Any:
return await self.send_request(url, method='GET', query=query)

Expand Down
32 changes: 20 additions & 12 deletions src/confluent_kafka/schema_registry/_sync/schema_registry_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,10 +97,10 @@ def __init__(
scope: str,
token_endpoint: str,
logical_cluster: str,
identity_pool: str,
max_retries: int,
retries_wait_ms: int,
retries_max_wait_ms: int,
identity_pool: Optional[str] = None,
):
self.token = None
self.logical_cluster = logical_cluster
Expand All @@ -113,11 +113,13 @@ def __init__(
self.token_expiry_threshold = 0.8

def get_bearer_fields(self) -> dict:
return {
fields = {
'bearer.auth.token': self.get_access_token(),
'bearer.auth.logical.cluster': self.logical_cluster,
'bearer.auth.identity.pool.id': self.identity_pool,
}
if self.identity_pool is not None:
fields['bearer.auth.identity.pool.id'] = self.identity_pool
return fields

def token_expired(self) -> bool:
if self.token is None:
Expand Down Expand Up @@ -283,19 +285,20 @@ def __init__(self, conf: dict):
self.auth = None

if self.bearer_auth_credentials_source in {'OAUTHBEARER', 'STATIC_TOKEN'}:
headers = ['bearer.auth.logical.cluster', 'bearer.auth.identity.pool.id']
missing_headers = [header for header in headers if header not in conf_copy]
if missing_headers:
if 'bearer.auth.logical.cluster' not in conf_copy:
raise ValueError(
"Missing required bearer configuration properties: {}".format(", ".join(missing_headers))
"Missing required bearer configuration property: bearer.auth.logical.cluster"
)

logical_cluster = conf_copy.pop('bearer.auth.logical.cluster')
if not isinstance(logical_cluster, str):
raise TypeError("logical cluster must be a str, not " + str(type(logical_cluster)))

identity_pool = conf_copy.pop('bearer.auth.identity.pool.id')
if not isinstance(identity_pool, str):
# Identity pool is optional and may be a single pool ID or comma-separated list of pool IDs.
# The value is passed as-is in the Confluent-Identity-Pool-Id header to Schema Registry.
# When union-of-pools is enabled on the server, multiple IDs are interpreted as a union.
identity_pool = conf_copy.pop('bearer.auth.identity.pool.id', None)
if identity_pool is not None and not isinstance(identity_pool, str):
raise TypeError("identity pool id must be a str, not " + str(type(identity_pool)))

if self.bearer_auth_credentials_source == 'OAUTHBEARER':
Expand Down Expand Up @@ -335,10 +338,10 @@ def __init__(self, conf: dict):
self.scope,
self.token_endpoint,
logical_cluster,
identity_pool,
self.max_retries,
self.retries_wait_ms,
self.retries_max_wait_ms,
identity_pool,
)
else: # STATIC_TOKEN
if 'bearer.auth.token' not in conf_copy:
Expand Down Expand Up @@ -412,7 +415,8 @@ def handle_bearer_auth(self, headers: dict) -> None:
if self.bearer_field_provider is None:
raise ValueError("Bearer field provider is not set")
bearer_fields = self.bearer_field_provider.get_bearer_fields()
required_fields = ['bearer.auth.token', 'bearer.auth.identity.pool.id', 'bearer.auth.logical.cluster']
# Note: bearer.auth.identity.pool.id is optional; only token and logical.cluster are required
required_fields = ['bearer.auth.token', 'bearer.auth.logical.cluster']

missing_fields = []
for field in required_fields:
Expand All @@ -427,9 +431,13 @@ def handle_bearer_auth(self, headers: dict) -> None:
)

headers["Authorization"] = "Bearer {}".format(bearer_fields['bearer.auth.token'])
headers['Confluent-Identity-Pool-Id'] = bearer_fields['bearer.auth.identity.pool.id']
headers['target-sr-cluster'] = bearer_fields['bearer.auth.logical.cluster']

# Identity pool may be a single pool ID or comma-separated list of pool IDs.
# Pass the value as-is to the server; it handles union-of-pools semantics.
if 'bearer.auth.identity.pool.id' in bearer_fields:
headers['Confluent-Identity-Pool-Id'] = bearer_fields['bearer.auth.identity.pool.id']

def get(self, url: str, query: Optional[dict] = None) -> Any:
return self.send_request(url, method='GET', query=query)

Expand Down
16 changes: 10 additions & 6 deletions src/confluent_kafka/schema_registry/common/schema_registry_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,33 +71,37 @@ async def get_bearer_fields(self) -> dict:
class _StaticFieldProvider(_BearerFieldProvider):
"""Synchronous static token bearer field provider."""

def __init__(self, token: str, logical_cluster: str, identity_pool: str):
def __init__(self, token: str, logical_cluster: str, identity_pool: Optional[str] = None):
self.token = token
self.logical_cluster = logical_cluster
self.identity_pool = identity_pool

def get_bearer_fields(self) -> dict:
return {
fields = {
'bearer.auth.token': self.token,
'bearer.auth.logical.cluster': self.logical_cluster,
'bearer.auth.identity.pool.id': self.identity_pool,
}
if self.identity_pool is not None:
fields['bearer.auth.identity.pool.id'] = self.identity_pool
return fields


class _AsyncStaticFieldProvider(_AsyncBearerFieldProvider):
"""Asynchronous static token bearer field provider."""

def __init__(self, token: str, logical_cluster: str, identity_pool: str):
def __init__(self, token: str, logical_cluster: str, identity_pool: Optional[str] = None):
self.token = token
self.logical_cluster = logical_cluster
self.identity_pool = identity_pool

async def get_bearer_fields(self) -> dict:
return {
fields = {
'bearer.auth.token': self.token,
'bearer.auth.logical.cluster': self.logical_cluster,
'bearer.auth.identity.pool.id': self.identity_pool,
}
if self.identity_pool is not None:
fields['bearer.auth.identity.pool.id'] = self.identity_pool
return fields


def is_success(status_code: int) -> bool:
Expand Down
Loading