ClassNotFound using SCRAM SHA 256 with Kafka extension 
## Describe the bug

Using the quarkus Kafka extension and using the following Kafka config:

```
kafka:
  bootstrap:
    servers: ...
  security:
    protocol: SASL_SSL
  sasl:
    mechanism: SCRAM-SHA-256
    jaas:
      config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${kafka.username}" password="${kafka.password}";
```

Works fine in JVM mode, but when compiling to native it does not work:

Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.security.scram.internals.ScramSaslClient$ScramSaslClientFactory
	at com.oracle.svm.core.hub.ClassForNameSupport.forName(ClassForNameSupport.java:64)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:290)
	at java.security.Provider$Service.getImplClass(Provider.java:1920)
	... 43 more


### Expected behavior
Being able to use SCAM SHA 256 authentication with Kafka in Quarkus.

### Actual behavior

From what I can see, some underlying Kafka client library classes are not included by GraalVM, and some may to work around that needs to be applied.


## To Reproduce

Use Kafka client with SCRAM SHA 256 auth against a Kafka cluster. 

### Configuration

Relevant parts:

```properties
kafka:
  bootstrap:
    servers: ...
  security:
    protocol: SASL_SSL
  sasl:
    mechanism: SCRAM-SHA-256
    jaas:
      config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${kafka.username}" password="${kafka.password}";
```

## Environment (please complete the following information):

### Output of `uname -a` or `ver`

Linux pteppic 5.12.10-300.fc34.x86_64 #1 SMP Thu Jun 10 14:21:36 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

### Output of `java -version`

OpenJDK 64-Bit Server VM 18.9 (build 11.0.11+9, mixed mode, sharing)

### GraalVM version (if different from Java)

graalvm-ce-java11-21.1.0

### Quarkus version or git rev

2.0.0.CR3

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f)

## Additional context

Full stack trace:

```
2021-06-20 19:00:25,119 INFO  [org.apa.kaf.com.net.SaslChannelBuilder] (smallrye-kafka-consumer-thread-0) [Consumer clientId=kafka-consumer-telemetry-inbound, groupId=plantmonitor] Failed to create channel due to : org.apache.kafka.common.errors.SaslAuthenticationException: Failed to configure SaslClientAuthenticator
Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Failed to create SaslClient with mechanism SCRAM-SHA-256
Caused by: javax.security.sasl.SaslException: Cannot instantiate service SASL/SCRAM Client Provider: SaslClientFactory.SCRAM-SHA-256 -> org.apache.kafka.common.security.scram.internals.ScramSaslClient$ScramSaslClientFactory
 [Caused by java.security.NoSuchAlgorithmException: class configured for SaslClientFactory (provider: SASL/SCRAM Client Provider) cannot be found.]
	at javax.security.sasl.Sasl.loadFactory(Sasl.java:460)
	at javax.security.sasl.Sasl.createSaslClient(Sasl.java:431)
	at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.lambda$createSaslClient$0(SaslClientAuthenticator.java:219)
	at java.security.AccessController.doPrivileged(AccessController.java:145)
	at javax.security.auth.Subject.doAs(Subject.java:36)
	at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:215)
	at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.<init>(SaslClientAuthenticator.java:206)
	at org.apache.kafka.common.network.SaslChannelBuilder.buildClientAuthenticator(SaslChannelBuilder.java:287)
	at org.apache.kafka.common.network.SaslChannelBuilder.lambda$buildChannel$1(SaslChannelBuilder.java:228)
	at org.apache.kafka.common.network.KafkaChannel.<init>(KafkaChannel.java:143)
	at org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:236)
	at org.apache.kafka.common.network.Selector.buildAndAttachKafkaChannel(Selector.java:338)
	at org.apache.kafka.common.network.Selector.registerChannel(Selector.java:329)
	at org.apache.kafka.common.network.Selector.connect(Selector.java:256)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:987)
	at org.apache.kafka.clients.NetworkClient.access$600(NetworkClient.java:73)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1158)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1046)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:559)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:227)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:164)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:257)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:480)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1261)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1230)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)
	at io.smallrye.reactive.messaging.kafka.impl.ReactiveKafkaConsumer.lambda$poll$4(ReactiveKafkaConsumer.java:127)
	at io.smallrye.reactive.messaging.kafka.impl.ReactiveKafkaConsumer.lambda$runOnPollingThread$0(ReactiveKafkaConsumer.java:96)
	at io.smallrye.context.impl.wrappers.SlowContextualSupplier.get(SlowContextualSupplier.java:21)
	at io.smallrye.mutiny.operators.uni.builders.UniCreateFromItemSupplier.subscribe(UniCreateFromItemSupplier.java:28)
	at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
	at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.lambda$subscribe$0(UniRunSubscribeOn.java:27)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.lang.Thread.run(Thread.java:829)
	at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:553)
	at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)
Caused by: java.security.NoSuchAlgorithmException: class configured for SaslClientFactory (provider: SASL/SCRAM Client Provider) cannot be found.
	at java.security.Provider$Service.getImplClass(Provider.java:1933)
	at java.security.Provider$Service.newInstance(Provider.java:1894)
	at javax.security.sasl.Sasl.loadFactory(Sasl.java:458)
	... 41 more
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.security.scram.internals.ScramSaslClient$ScramSaslClientFactory
	at com.oracle.svm.core.hub.ClassForNameSupport.forName(ClassForNameSupport.java:64)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:290)
	at java.security.Provider$Service.getImplClass(Provider.java:1920)
	... 43 more
```
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 5e797c4b83c..b1be4d68871 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
@@ -18,12 +18,14 @@
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.authenticator.AbstractLogin;
 import org.apache.kafka.common.security.authenticator.DefaultLogin;
 import org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
 import org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin;
 import org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClient;
+import org.apache.kafka.common.security.scram.internals.ScramSaslClient;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.ByteBufferDeserializer;
@@ -44,6 +46,7 @@
 import org.apache.kafka.common.serialization.ShortSerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.eclipse.microprofile.config.ConfigProvider;
 import org.jboss.jandex.ClassInfo;
 import org.jboss.jandex.DotName;
 import org.jboss.jandex.Type;
@@ -408,12 +411,21 @@ public AdditionalBeanBuildItem runtimeConfig() {
 
     @BuildStep
     public void withSasl(BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
-            BuildProducer<ReflectiveHierarchyBuildItem> reflectiveHierarchy) {
+            BuildProducer<ReflectiveHierarchyBuildItem> reflectiveHierarchy,
+            BuildProducer<ExtensionSslNativeSupportBuildItem> sslNativeSupport) {
 
         reflectiveClass
                 .produce(new ReflectiveClassBuildItem(false, false, AbstractLogin.DefaultLoginCallbackHandler.class));
         reflectiveClass.produce(new ReflectiveClassBuildItem(false, false, SaslClientCallbackHandler.class));
         reflectiveClass.produce(new ReflectiveClassBuildItem(false, false, DefaultLogin.class));
+        reflectiveClass
+                .produce(new ReflectiveClassBuildItem(true, false, false, ScramSaslClient.ScramSaslClientFactory.class));
+
+        // Enable SSL support if kafka.security.protocol is set to something other than PLAINTEXT, which is the default
+        String securityProtocol = ConfigProvider.getConfig().getConfigValue("kafka.security.protocol").getValue();
+        if (securityProtocol != null && SecurityProtocol.forName(securityProtocol) != SecurityProtocol.PLAINTEXT) {
+            sslNativeSupport.produce(new ExtensionSslNativeSupportBuildItem(Feature.KAFKA_CLIENT));
+        }
 
         final Type loginModuleType = Type
                 .create(DotName.createSimple(LoginModule.class.getName()), Kind.CLASS);
['extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java']