KafkaConsumer generic types for avro SpecificRecord using Reflection

user_10915278 0
Viewed 317

I want to initialize KafkaConsumer generic types using classes loaded with reflection. The classes will be generated via avro-tools and will be extending SpecificRecordBase class. I know, I can use wildcards and do something like -

KafkaConsumer<? extends SpecificRecordBase, ? extends SpecificRecordBase> consumer = new KafkaConsumer<>(properties);

However, I am looking for something like -

Class keyClass = Class.forName("com.test.KeyClass");
Class valueClass = Class.forName("com.test.ValueClass");
KafkaConsumer<keyClass, valueClass> consumer = new KafkaConsumer<>(properties);

Any pointers would be appreciated. Thanks.

apache-kafka avro reflection java
3 Answers

Since type parameters exist only for programmers convenience, in runtime they will be erased, thus, it makes no sense to load them by reflection.

However, you can cast it to needed type, e.g.

Consumer<KeyClass, ValueClass> consumer = new KafkaConsumer<>(properties);
user_51489210

I am not sure but if below suffice your requirement --

public Consumer<?,?> getConsumer(Class<?> keyClass,Class<?> valClass,
                                   Properties props){
    props.put(deserializer properties of keyClass);
    props.put(deserializer properties of valClass);
   //other releavant props 
   return new KafkaConsumer<>(props);
}
  ..
   caller(){
   Class<?> keyClass=Class.forName("KeyClass");
   Class<?> valClass=Class.forName("ValClass");
   KafkaConsumer<?,?> con=(KafkaConsumer<?,?>)getConsumer(keyClass,valClass,props);
}
user_50142210

Setup a separate task to build and publish the schema jar (I'd recommend putting them in a separate repo with it's own build). Then pull in the dependency in through Gradle/Maven, you'll have the actual types and reflection shouldn't be necessary.

user_17351460
Related