spring.kafka.consumer.properties.spring.json.value.type.method=com.example.demo.MyClass.returnType

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface KafkaMessage {

    String[] typeId() default "";

}

public class KafkaMessageAnnotationPostProcessor implements BeanPostProcessor {

    private static final String SCAN_PACKAGE_PROPERTY_KEY = "spring.kafka.message.package";
    private static final String DEFAULT_SCAN_PACKAGE = "*";

    private final Environment environment;

    public KafkaMessageAnnotationPostProcessor(Environment environment) {
        this.environment = environment;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (JsonMessageConverter.class.isAssignableFrom(bean.getClass())) {
            setTypeMapper((JsonMessageConverter)bean, getMapping());
        }
        return bean;
    }

    private Map<String, Class<?>> getMapping() {
        Map<String, Class<?>> mapping = new HashMap<>();
        for (BeanDefinition beanDefinition : findAnnotatedClasses(getScanPackage())) {
            addToMapping(mapping, getClass(beanDefinition));
        }
        return mapping;
    }

    private String getScanPackage() {
        return environment.getProperty(SCAN_PACKAGE_PROPERTY_KEY, DEFAULT_SCAN_PACKAGE);
    }

    private Set<BeanDefinition> findAnnotatedClasses(String scanPackage) {
        ClassPathScanningCandidateComponentProvider provider = createComponentProvider();
        return provider.findCandidateComponents(scanPackage);
    }

    private ClassPathScanningCandidateComponentProvider createComponentProvider() {
        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
        provider.addIncludeFilter(new AnnotationTypeFilter(KafkaMessage.class));
        return provider;
    }

    private Class<?> getClass(BeanDefinition beanDefinition) {
        try {
            return Class.forName(beanDefinition.getBeanClassName());
        }
        catch (ClassNotFoundException e) {
            throw new IllegalStateException("Cannot find " + beanDefinition.getBeanClassName(), e);
        }
    }

    private void addToMapping(Map<String, Class<?>> mapping, Class<?> clazz) {
        for (String i : getTypeIds(clazz)) {
            mapping.put(i, clazz);
        }
    }

    private List<String> getTypeIds(Class<?> clazz) {
        KafkaMessage annotation = clazz.getAnnotation(KafkaMessage.class);
        // Use simple name as default
        return annotation.typeId().length > 0 ? Arrays.asList(annotation.typeId()) : Collections.singletonList(clazz.getSimpleName());
    }

    private void setTypeMapper(JsonMessageConverter converter, Map<String, Class<?>> mapping) {
        // Only set a mapping if an @KafkaMessage annotation is set on any class
        if (!mapping.isEmpty()) {
            Jackson2JavaTypeMapper typeMapper = converter.getTypeMapper();
            if (!AbstractJavaTypeMapper.class.isAssignableFrom(typeMapper.getClass())) {
                throw new IllegalStateException("TypeMapper must be of type AbstractJavaTypeMapper");
            }
            typeMapper.setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
            ((AbstractJavaTypeMapper)typeMapper).setIdClassMapping(mapping);
        }
    }

}