Repository: quarkusio/quarkus
Issue: Kafka Streams Extension can't find Serdes.StringSerde

**Describe the bug**
Native image generated cannot find the StringSerde. This causes the app to fail on startup.

**Expected behavior**
The app should start up and run without issues.

**Actual behavior**
I have a Java Kafka Streams app. I configure the Serdes.StringSerde as the default Serdes in the Kafka properties.
I then build a native image (either on OSX or docker).
When I run the native image, it fails with:
```
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.Serdes$StringSerde for configuration default.key.serde: Class org.apache.kafka.common.serialization.Serdes$StringSerde could not be found.
	at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:718)
	at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:471)
	at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:464)
	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:62)
	at org.apache.kafka.streams.StreamsConfig.<init>(StreamsConfig.java:849)
	at org.apache.kafka.streams.StreamsConfig.<init>(StreamsConfig.java:844)
	at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:544)
	at com.aimyourtechnology.quarkus.kafka.streams.ConverterStream.runTopology(ConverterStream.java:52)
	at com.aimyourtechnology.quarkus.kafka.streams.StreamingApp.onStart(StreamingApp.java:54)
	at com.aimyourtechnology.quarkus.kafka.streams.StreamingApp_Observer_onStart_fd71b5e0b207b7d1ef838b94eaeff75e52b8f463.notify(StreamingApp_Observer_onStart_fd71b5e0b207b7d1ef838b94eaeff75e52b8f463.zig:63)
	at io.quarkus.arc.EventImpl$Notifier.notify(EventImpl.java:228)
	at io.quarkus.arc.EventImpl.fire(EventImpl.java:69)
	at io.quarkus.arc.runtime.LifecycleEventRunner.fireStartupEvent(LifecycleEventRunner.java:23)
	at io.quarkus.arc.runtime.ArcDeploymentTemplate.handleLifecycleEvents(ArcDeploymentTemplate.java:99)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent17.deploy_0(LifecycleEventsBuildStep$startupEvent17.zig:58)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent17.deploy(LifecycleEventsBuildStep$startupEvent17.zig:77)
	at io.quarkus.runner.ApplicationImpl1.doStart(ApplicationImpl1.zig:90)
	... 3 more
```


**To Reproduce**
Steps to reproduce the behavior:
You can show it through docker:
```
1. git clone https://github.com/muirandy/quarkus-kafka-streams-xml-json-converter.git
2. cd quarkus-kafka-streams-xml-json-converter
3. mvn package -Pnative -Dnative-image.docker-build=true
4. docker build -f src/main/docker/Dockerfile.native -t quarkus/quarkus-kafka-streams-xml-json-converter .
```
You can now run acceptance.converter.ConvertToJsonShould from an IDE. This will demonstrate the failure.


**Environment (please complete the following information):**
 - Output of `uname -a` or `ver`: 
```
Darwin andymuir-3.local 18.6.0 Darwin Kernel Version 18.6.0: Thu Apr 25 23:16:27 PDT 2019; root:xnu-4903.261.4~2/RELEASE_X86_64 x86_64
```
 - Output of `java -version`: 
```
java version "1.8.0_212"
Java(TM) SE Runtime Environment (build 1.8.0_212-b31)
Java HotSpot(TM) 64-Bit GraalVM EE 19.1.0 (build 25.212-b31-jvmci-20-b04, mixed mode)
```
 - GraalVM version (if different from Java): 19.1.0
 - Quarkus version or git rev: 0.18.0

**Additional context**
The quickstart example uses a StringSerde defined for the Stream:
```
builder.stream(                                                       
                        TEMPERATURE_VALUES_TOPIC,
                        Consumed.with(Serdes.Integer(), Serdes.String())
                )
```

My code defines:
```
        streamingConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        streamingConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
```
as part of the Kafka Properties used when creating the KafkaStreams within com.aimyourtechnology.quarkus.kafka.streams.ConverterStream:
```
streams = new KafkaStreams(topology, streamingConfig);
```


Changed files:
extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java

diff --git a/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
index 9b8cf410880..c46fd119b9b 100644
--- a/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
+++ b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
@@ -4,6 +4,7 @@
 import java.util.Properties;
 
 import org.apache.kafka.common.serialization.Serdes.ByteArraySerde;
+import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
 import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
 import org.apache.kafka.streams.processor.DefaultPartitionGrouper;
@@ -44,13 +45,62 @@ void build(RecorderContext recorder,
 
         feature.produce(new FeatureBuildItem(FeatureBuildItem.KAFKA_STREAMS));
 
+        registerClassesThatAreLoadedThroughReflection(reflectiveClasses);
+        addSupportForRocksDbLib(nativeLibs);
+        enableLoadOfNativeLibs(reinitialized);
+        enableJniForNativeBuild(jni);
+    }
+
+    private void registerClassesThatAreLoadedThroughReflection(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        registerCompulsoryClasses(reflectiveClasses);
+        registerClassesThatClientMaySpecify(reflectiveClasses);
+    }
+
+    private void registerCompulsoryClasses(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, StreamsPartitionAssignor.class));
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultPartitionGrouper.class));
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultProductionExceptionHandler.class));
-        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, LogAndFailExceptionHandler.class));
-        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, ByteArraySerde.class));
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, FailOnInvalidTimestamp.class));
+    }
+
+    private void registerClassesThatClientMaySpecify(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        registerExceptionHandler(reflectiveClasses);
+        registerDefaultSerdes(reflectiveClasses);
+    }
+
+    private void registerExceptionHandler(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        Properties properties = buildKafkaStreamsProperties();
+        String exceptionHandlerClassName = properties
+                .getProperty(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG);
 
+        if (exceptionHandlerClassName == null) {
+            registerDefaultExceptionHandler(reflectiveClasses);
+        } else {
+            registerClassName(reflectiveClasses, exceptionHandlerClassName);
+        }
+    }
+
+    private void registerDefaultExceptionHandler(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, LogAndFailExceptionHandler.class));
+    }
+
+    private void registerDefaultSerdes(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        Properties properties = buildKafkaStreamsProperties();
+        String defaultKeySerdeClass = properties.getProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG);
+        String defaultValueSerdeClass = properties.getProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG);
+
+        if (defaultKeySerdeClass != null) {
+            registerClassName(reflectiveClasses, defaultKeySerdeClass);
+        }
+        if (defaultValueSerdeClass != null) {
+            registerClassName(reflectiveClasses, defaultValueSerdeClass);
+        }
+        if (!allDefaultSerdesAreDefinedInProperties(defaultKeySerdeClass, defaultValueSerdeClass)) {
+            registerDefaultSerde(reflectiveClasses);
+        }
+    }
+
+    private void addSupportForRocksDbLib(BuildProducer<SubstrateResourceBuildItem> nativeLibs) {
         // for RocksDB, either add linux64 native lib when targeting containers
         if (isContainerBuild()) {
             nativeLibs.produce(new SubstrateResourceBuildItem("librocksdbjni-linux64.so"));
@@ -59,27 +109,53 @@ void build(RecorderContext recorder,
         else {
             nativeLibs.produce(new SubstrateResourceBuildItem(Environment.getJniLibraryFileName("rocksdb")));
         }
+    }
 
-        // re-initializing RocksDB to enable load of native libs
+    private void enableLoadOfNativeLibs(BuildProducer<RuntimeReinitializedClassBuildItem> reinitialized) {
         reinitialized.produce(new RuntimeReinitializedClassBuildItem("org.rocksdb.RocksDB"));
+    }
 
+    private void enableJniForNativeBuild(BuildProducer<JniBuildItem> jni) {
         jni.produce(new JniBuildItem());
     }
 
+    private void registerClassName(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses, String defaultKeySerdeClass) {
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, defaultKeySerdeClass));
+    }
+
+    private boolean allDefaultSerdesAreDefinedInProperties(String defaultKeySerdeClass, String defaultValueSerdeClass) {
+        return defaultKeySerdeClass != null && defaultValueSerdeClass != null;
+    }
+
+    private void registerDefaultSerde(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, ByteArraySerde.class));
+    }
+
     @BuildStep
     @Record(ExecutionTime.STATIC_INIT)
     BeanContainerListenerBuildItem processBuildTimeConfig(KafkaStreamsRecorder recorder) {
-        Config config = ConfigProvider.getConfig();
+        Properties kafkaStreamsProperties = buildKafkaStreamsProperties();
+        return new BeanContainerListenerBuildItem(recorder.configure(kafkaStreamsProperties));
+    }
 
-        Properties properties = new Properties();
+    private Properties buildKafkaStreamsProperties() {
+        Config config = ConfigProvider.getConfig();
+        Properties kafkaStreamsProperties = new Properties();
         for (String property : config.getPropertyNames()) {
-            if (property.startsWith(STREAMS_OPTION_PREFIX)) {
-                properties.setProperty(property.substring(STREAMS_OPTION_PREFIX.length()),
-                        config.getValue(property, String.class));
+            if (isKafkaStreamsProperty(property)) {
+                includeKafkaStreamsProperty(config, kafkaStreamsProperties, property);
             }
         }
+        return kafkaStreamsProperties;
+    }
+
+    private boolean isKafkaStreamsProperty(String property) {
+        return property.startsWith(STREAMS_OPTION_PREFIX);
+    }
 
-        return new BeanContainerListenerBuildItem(recorder.configure(properties));
+    private void includeKafkaStreamsProperty(Config config, Properties kafkaStreamsProperties, String property) {
+        kafkaStreamsProperties.setProperty(property.substring(STREAMS_OPTION_PREFIX.length()),
+                config.getValue(property, String.class));
     }
 
     @BuildStep