KafkaConnect with schema registry (apicurio) oauth Azure AD

[seb2020]

Hi,

I am running a KafkaConnector that need to access the schema registry with an oath authentication trough Azure AD but it's not working. The use case is with Debezium.

I am running :

• Streams for Apache Kafka 2.7.0-7
• Red Hat build of Apicurio Registry 2.6.8+0.1739457262.p

The KafkaConnect configuration :

key.converter: io.apicurio.registry.utils.converter.AvroConverter
key.converter.apicurio.registry.url: https://apicurio-registry.apps.openshit.tld/apis/registry/v2
key.converter.apicurio.rest.request.ssl.ca-bundle.location: /opt/kafka/external-configuration/ca-bundle/ca-certificates.crt
key.converter.apicurio.registry.auto-register: true
key.converter.apicurio.registry.find-latest: true
key.converter.apicurio.registry.auth.service.token.endpoint: https://login.microsoftonline.com/<tenant>/oauth2/v2.0/token/
key.converter.apicurio.registry.auth.client.id: <id>
key.converter.apicurio.registry.auth.client.secret: <secret>
key.converter.apicurio.registry.auth.client.scope: <id>/.default

The error in the KafkaConnect pod :

2025-02-27 07:26:26,466 ERROR [kafka-connector-ulysse-int-postgres-apicurio|task-0] WorkerSourceTask{id=kafka-connector-ulysse-int-postgres-apicurio-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-kafka-connector-ulysse-int-postgres-apicurio-0]
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:230)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:156)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.convertTransformedRecord(AbstractWorkerSourceTask.java:491)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.sendRecords(AbstractWorkerSourceTask.java:402)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:367)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:77)
    at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:237)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: io.apicurio.rest.client.auth.exception.NotAuthorizedException: Authentication exception
    at io.apicurio.registry.rest.client.impl.ErrorHandler.handleErrorResponse(ErrorHandler.java:54)
    at io.apicurio.rest.client.handler.BodyHandler.lambda$toSupplierOfType$1(BodyHandler.java:55)
    at io.apicurio.rest.client.JdkHttpClient.sendRequest(JdkHttpClient.java:204)
    at io.apicurio.registry.rest.client.impl.RegistryClientImpl.createArtifact(RegistryClientImpl.java:311)
    at io.apicurio.registry.rest.client.RegistryClient.createArtifact(RegistryClient.java:147)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.lambda$handleAutoCreateArtifact$4(DefaultSchemaResolver.java:280)
    at io.apicurio.registry.resolver.ERCache.lambda$getValue$0(ERCache.java:201)
    at io.apicurio.registry.resolver.ERCache.retry(ERCache.java:254)
    at io.apicurio.registry.resolver.ERCache.getValue(ERCache.java:200)
    at io.apicurio.registry.resolver.ERCache.getByContent(ERCache.java:175)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.handleAutoCreateArtifact(DefaultSchemaResolver.java:278)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.getSchemaFromRegistry(DefaultSchemaResolver.java:125)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.lambda$resolveSchema$0(DefaultSchemaResolver.java:93)
    at java.base/java.util.Optional.orElseGet(Optional.java:364)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.resolveSchema(DefaultSchemaResolver.java:93)
    at io.apicurio.registry.serde.AbstractKafkaSerializer.serialize(AbstractKafkaSerializer.java:83)
    at io.apicurio.registry.utils.converter.SerdeBasedConverter.fromConnectData(SerdeBasedConverter.java:122)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.lambda$convertTransformedRecord$5(AbstractWorkerSourceTask.java:491)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214)
    ... 13 more

On the schema registry, I have this :

2025-02-25 06:16:13 DEBUG <> [io.quarkus.oidc.runtime.OidcAuthenticationMechanism$3] (executor-thread-196) Resolved OIDC tenant id: Default
2025-02-25 06:16:13 DEBUG <> [io.quarkus.oidc.runtime.BearerAuthenticationMechanism] (executor-thread-196) Starting a bearer access token authentication
2025-02-25 06:16:13 DEBUG <> [io.quarkus.oidc.runtime.OidcUtils] (executor-thread-196) Looking for a token in the authorization header
2025-02-25 06:16:13 DEBUG <> [io.quarkus.oidc.runtime.BearerAuthenticationMechanism] (executor-thread-196) Bearer access token is not available

So for me, the KafkaConnector doesn't do any authentication because I cannot see a connection to Azure AD and therefore the schema registry didn't see the correct header.

Is this a bug or did I make an error in the configuration ?

[ppatierno]

AFAICS you are using the Red Hat products so you should refer to the proper support.
Other than that, it seems to be an issue between Debezium and Apicurio so this would not be the right place to raise it.