GH-4110: AVRO related AOT reflection

Fixes #4110

Refactor `KafkaAvroBeanRegistrationAotProcessor` to use `BindingReflectionHintsRegistrar`

Replace manual reflection hints registration with `BindingReflectionHintsRegistrar`
to simplify AVRO type reflection configuration in AOT processing.

Adding `org.apache.kafka.common.security.oauthbearer.DefaultJwtRetriever` to Kafka runtime hints.

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

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

@@ -26,7 +26,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.jspecify.annotations.Nullable;
 
-import org.springframework.aot.hint.MemberCategory;
+import org.springframework.aot.hint.BindingReflectionHintsRegistrar;
 import org.springframework.aot.hint.ReflectionHints;
 import org.springframework.beans.factory.aot.BeanRegistrationAotContribution;
 import org.springframework.beans.factory.aot.BeanRegistrationAotProcessor;
@@ -42,6 +42,7 @@
  *
  * @author Gary Russell
  * @author Sagnhyeok An
+ * @author Soby Chacko
  * @since 3.0
  *
  */
@@ -55,6 +56,8 @@ public class KafkaAvroBeanRegistrationAotProcessor implements BeanRegistrationAo
 
 	private static final boolean AVRO_PRESENT = ClassUtils.isPresent(AVRO_GENERATED_CLASS_NAME, null);
 
+	private final BindingReflectionHintsRegistrar bindingRegistrar = new BindingReflectionHintsRegistrar();
+
 	@Override
 	public @Nullable BeanRegistrationAotContribution processAheadOfTime(RegisteredBean registeredBean) {
 		if (!AVRO_PRESENT) {
@@ -83,9 +86,7 @@ public class KafkaAvroBeanRegistrationAotProcessor implements BeanRegistrationAo
 		if (!avroTypes.isEmpty()) {
 			return (generationContext, beanRegistrationCode) -> {
 				ReflectionHints reflectionHints = generationContext.getRuntimeHints().reflection();
-				avroTypes.forEach(type -> reflectionHints.registerType(type,
-						builder -> builder.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS,
-								MemberCategory.INVOKE_PUBLIC_METHODS)));
+				this.bindingRegistrar.registerReflectionHints(reflectionHints, avroTypes.toArray(new Class<?>[0]));
 			};
 		}
 		return null;

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

@@ -24,6 +24,7 @@
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
 import org.apache.kafka.common.protocol.Message;
+import org.apache.kafka.common.security.oauthbearer.DefaultJwtRetriever;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.AppInfoParser.AppInfo;
 import org.apache.kafka.common.utils.ImplicitLinkedHashCollection;
@@ -87,6 +88,8 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 						builder -> builder.withMembers(MemberCategory.INVOKE_DECLARED_METHODS)));
 
 		Stream.of(
+						// Following Kafka classes need to be ideally part of Oracle's reachability metadata repository.
+						DefaultJwtRetriever.class,
 						Message.class,
 						ImplicitLinkedHashCollection.Element.class,
 						NewTopic.class,