Remove org.apache.kafka native runtime hints

* Since several of the native runtime hints that Spring for Apache Kafka
  provides for org.apahce.kafka classes are available via the graalvm-reachability-metadata
  repository (https://github.com/oracle/graalvm-reachability-metadata), we are no longer
  keeping these runtime hints in Spring for Apache Kafka since version 3.2.0. This applies
  to both kafka-client classes and Kafka Streams classes.

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/aot/KafkaRuntimeHints.java

@@ -21,33 +21,10 @@
 
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
-import org.apache.kafka.clients.consumer.RangeAssignor;
-import org.apache.kafka.clients.consumer.RoundRobinAssignor;
-import org.apache.kafka.clients.consumer.StickyAssignor;
 import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.RoundRobinPartitioner;
 import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
 import org.apache.kafka.common.protocol.Message;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.ByteBufferDeserializer;
-import org.apache.kafka.common.serialization.ByteBufferSerializer;
-import org.apache.kafka.common.serialization.BytesDeserializer;
-import org.apache.kafka.common.serialization.BytesSerializer;
-import org.apache.kafka.common.serialization.DoubleDeserializer;
-import org.apache.kafka.common.serialization.DoubleSerializer;
-import org.apache.kafka.common.serialization.FloatDeserializer;
-import org.apache.kafka.common.serialization.FloatSerializer;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.ListDeserializer;
-import org.apache.kafka.common.serialization.ListSerializer;
-import org.apache.kafka.common.serialization.LongDeserializer;
-import org.apache.kafka.common.serialization.LongSerializer;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.AppInfoParser.AppInfo;
 import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
 
@@ -140,34 +117,10 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 
 		Stream.of(
 					AppInfo.class,
-					// standard assignors
-					CooperativeStickyAssignor.class,
-					RangeAssignor.class,
-					RoundRobinAssignor.class,
-					StickyAssignor.class,
 					// standard partitioners
 					org.apache.kafka.clients.producer.internals.DefaultPartitioner.class,
-					RoundRobinPartitioner.class,
 					org.apache.kafka.clients.producer.UniformStickyPartitioner.class,
 					// standard serialization
-					ByteArrayDeserializer.class,
-					ByteArraySerializer.class,
-					ByteBufferDeserializer.class,
-					ByteBufferSerializer.class,
-					BytesDeserializer.class,
-					BytesSerializer.class,
-					DoubleSerializer.class,
-					DoubleDeserializer.class,
-					FloatSerializer.class,
-					FloatDeserializer.class,
-					IntegerSerializer.class,
-					IntegerDeserializer.class,
-					ListDeserializer.class,
-					ListSerializer.class,
-					LongSerializer.class,
-					LongDeserializer.class,
-					StringDeserializer.class,
-					StringSerializer.class,
 					// Spring serialization
 					DelegatingByTopicDeserializer.class,
 					DelegatingByTypeSerializer.class,
@@ -180,17 +133,6 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 					StringOrBytesSerializer.class,
 					ToStringSerializer.class,
 					Serdes.class,
-					Serdes.ByteArraySerde.class,
-					Serdes.BytesSerde.class,
-					Serdes.ByteBufferSerde.class,
-					Serdes.DoubleSerde.class,
-					Serdes.FloatSerde.class,
-					Serdes.IntegerSerde.class,
-					Serdes.LongSerde.class,
-					Serdes.ShortSerde.class,
-					Serdes.StringSerde.class,
-					Serdes.UUIDSerde.class,
-					Serdes.VoidSerde.class,
 					CRC32C.class)
 				.forEach(type -> reflectionHints.registerType(type, builder ->
 						builder.withMembers(MemberCategory.INVOKE_PUBLIC_CONSTRUCTORS)));
@@ -200,15 +142,10 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 
 		Stream.of(
 				"sun.security.provider.ConfigFile",
-				"org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor",
-				"org.apache.kafka.streams.errors.DefaultProductionExceptionHandler",
-				"org.apache.kafka.streams.processor.FailOnInvalidTimestamp",
-				"org.apache.kafka.streams.processor.internals.assignment.HighAvailabilityTaskAssignor",
 				"org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor",
 				"org.apache.kafka.streams.processor.internals.assignment.FallbackPriorTaskAssignor",
 				"org.apache.kafka.streams.state.BuiltInDslStoreSuppliers$RocksDBDslStoreSuppliers",
-				"org.apache.kafka.streams.state.BuiltInDslStoreSuppliers$InMemoryDslStoreSuppliers",
-				"org.apache.kafka.streams.errors.LogAndFailExceptionHandler")
+				"org.apache.kafka.streams.state.BuiltInDslStoreSuppliers$InMemoryDslStoreSuppliers")
 			.forEach(type -> reflectionHints.registerTypeIfPresent(classLoader, type, builder ->
 					builder.withMembers(MemberCategory.INVOKE_PUBLIC_CONSTRUCTORS)));
 	}