Add support for custom OAuth functions

[Claimundefine]

What

Add custom OAuth implementation. Created new base class for bearer field provider, which will be called to retrieve all fields that need to be modified in the header.

Checklist

• Contains customer facing changes? Including API/behavior changes
• Did you add sufficient unit test and/or integration test coverage for this PR?
  • If not, please explain why it is not required

References

JIRA:

https://confluentinc.atlassian.net/browse/DGS-17473

Test & Review

Tested locally with OAuth for custom.

Open questions / Follow-ups

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

PR Overview

This PR adds support for custom OAuth functionality by introducing a common _BearerFieldProvider interface with implementations for OAuth, static tokens, and custom logic. It also updates existing tests and configuration handling to work with the new provider abstraction while removing legacy OAuth client tests.

Reviewed Changes

File	Description
tests/schema_registry/test_bearer_field_provider.py	Adds tests covering expiration, token retrieval, static and custom OAuth implementations.
tests/schema_registry/test_config.py	Adds tests to validate custom bearer configuration handling.
src/confluent_kafka/schema_registry/schema_registry_client.py	Refactors bearer auth handling by introducing _BearerFieldProvider and related classes, and adapts configuration accordingly.
tests/schema_registry/test_oauth_client.py	Removes legacy OAuth client tests in favor of the new provider approach.

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (2)

tests/schema_registry/test_bearer_field_provider.py:95

• [nitpick] The test for the custom OAuth client only asserts that get_bearer_fields() returns the same value on successive calls. Consider asserting specific expected fields from the custom function to improve test coverage.

def test_custom_oauth_client():

src/confluent_kafka/schema_registry/schema_registry_client.py:84

• [nitpick] Consider renaming '_OAuthClient' to 'OAuthBearerFieldProvider' to better reflect its role as an implementation of _BearerFieldProvider for OAuth logic.

class _OAuthClient(_BearerFieldProvider):

[Elmerboul]

Hi, a potential user of this functionality here 😁 Do you already have a (rough) estimate on when you plan to release this?

[rayokota]

Thanks @Claimundefine , left some nits, otherwise looking good

[Claimundefine]

Hi, a potential user of this functionality here 😁 Do you already have a (rough) estimate on when you plan to release this?

HI @Elmerboul, the projected release date is end of March.

[sonarqube-confluent]

Analysis Details

6 Issues

• 0 Bugs
• 1 Vulnerability
• 5 Code Smells

Coverage and Duplications

• 84.10% Coverage (62.10% Estimated after merge)
• No duplication information (0.90% Estimated after merge)

Project ID: confluent-kafka-python

View in SonarQube

[rayokota]

Thanks @Claimundefine , LGTM