From 6d06f4f41d43fab6772176328d1a3d08b2a983ca Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Szynkiewicz?= <michal.l.szynkiewicz@gmail.com>
Date: Tue, 9 Mar 2021 08:23:28 +0100
Subject: [PATCH] ServiceProvider registration for Confluent Schema Registry
 basic auth

fixes #15448
---
 .../client/deployment/KafkaProcessor.java     | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
index 52886b8f8066c..8de48535f98ac 100644
--- a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
+++ b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
@@ -7,6 +7,7 @@
 
 import javax.security.auth.spi.LoginModule;
 
+import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
 import org.apache.kafka.clients.consumer.ConsumerInterceptor;
 import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
 import org.apache.kafka.clients.consumer.RangeAssignor;
@@ -117,6 +118,7 @@ void contributeClassesToIndex(BuildProducer<AdditionalIndexedClassesBuildItem> a
     public void build(
             KafkaBuildTimeConfig config,
             CombinedIndexBuildItem indexBuildItem, BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
+            BuildProducer<ServiceProviderBuildItem> serviceProviders,
             BuildProducer<NativeImageProxyDefinitionBuildItem> proxies,
             Capabilities capabilities, BuildProducer<UnremovableBeanBuildItem> beans,
             BuildProducer<NativeImageResourceBuildItem> nativeLibs, NativeConfig nativeConfig) {
@@ -172,7 +174,7 @@ public void build(
         reflectiveClass.produce(new ReflectiveClassBuildItem(true, false, "java.nio.DirectByteBuffer"));
         reflectiveClass.produce(new ReflectiveClassBuildItem(true, false, "sun.misc.Cleaner"));
 
-        handleAvro(reflectiveClass, proxies);
+        handleAvro(reflectiveClass, proxies, serviceProviders);
         handleOpenTracing(reflectiveClass, capabilities);
         handleStrimziOAuth(reflectiveClass);
         if (config.snappyEnabled) {
@@ -253,7 +255,8 @@ private void handleStrimziOAuth(BuildProducer<ReflectiveClassBuildItem> reflecti
     }
 
     private void handleAvro(BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
-            BuildProducer<NativeImageProxyDefinitionBuildItem> proxies) {
+            BuildProducer<NativeImageProxyDefinitionBuildItem> proxies,
+            BuildProducer<ServiceProviderBuildItem> serviceProviders) {
         // Avro - for both Confluent and Apicurio
         try {
             Class.forName("io.confluent.kafka.serializers.KafkaAvroDeserializer", false,
@@ -292,6 +295,18 @@ private void handleAvro(BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
             //ignore, Confluent Avro is not in the classpath
         }
 
+        try {
+            Class.forName("io.confluent.kafka.schemaregistry.client.security.basicauth.BasicAuthCredentialProvider", false,
+                    Thread.currentThread().getContextClassLoader());
+            serviceProviders
+                    .produce(new ServiceProviderBuildItem(
+                            "io.confluent.kafka.schemaregistry.client.security.basicauth.BasicAuthCredentialProvider",
+                            "io.confluent.kafka.schemaregistry.client.security.basicauth.SaslBasicAuthCredentialProvider",
+                            "io.confluent.kafka.schemaregistry.client.security.basicauth.UrlBasicAuthCredentialProvider",
+                            "io.confluent.kafka.schemaregistry.client.security.basicauth.UserInfoCredentialProvider"));
+        } catch (ClassNotFoundException e) {
+            // ignore, Confluent schema registry client not in the classpath
+        }
         try {
             Class.forName("io.apicurio.registry.utils.serde.AvroKafkaDeserializer", false,
                     Thread.currentThread().getContextClassLoader());