@KafkaListner NOT Able to Fire/Trigger - For Dynamic Component about poc-spring-boot-dynamic-controller HOT 10 OPEN

package com.example;

import com.example.MyPayload;
import java.lang.reflect.Modifier;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.asm.MemberAttributeExtension;
import net.bytebuddy.description.annotation.AnnotationDescription;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.Argument;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.context.annotation.Configuration;

import static net.bytebuddy.matcher.ElementMatchers.named;

@Component
public class MyBeanFactoryPostProcessor implements BeanFactoryPostProcessor {

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        try {
            // registers dynamicKafkaListenerConfig bean
            Class<?> dynamicKafkaListenerConfig = generateDynamicKafkaListenerConfig();
            AbstractBeanDefinition dynamicKafkaListenerConfigBeanDefinition = BeanDefinitionBuilder
                .rootBeanDefinition(dynamicKafkaListenerConfig)
                .getBeanDefinition();
            ((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(
                "dynamicKafkaListenerConfig", dynamicKafkaListenerConfigBeanDefinition);

            // registers dynamicKafkaListener bean
            Class<?> dynamicKafkaListener = generateDynamicKafkaListener();
            AbstractBeanDefinition dynamicKafkaListenerBeanDefinition = BeanDefinitionBuilder
                .rootBeanDefinition(dynamicKafkaListener)
                .getBeanDefinition();

            ((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(
                "dynamicKafkaListener", dynamicKafkaListenerBeanDefinition);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    // generates dynamicKafkaListenerConfig class
    private Class<?> generateDynamicKafkaListenerConfig() throws Exception {
        return new ByteBuddy()

            .subclass(Object.class)
            .name("DynamicKafkaListenerConfig")
            .annotateType(AnnotationDescription.Builder.ofType(Configuration.class)
                .build(), AnnotationDescription.Builder.ofType(EnableKafka.class)
                .build())

            .make()
            .load(getClass().getClassLoader())
            .getLoaded();
    }

    // generates dynamicKafkaListener class
    private Class<?> generateDynamicKafkaListener() throws Exception {
        return new ByteBuddy()

            .subclass(Object.class)
            .name("DynamicKafkaListener")
            .annotateType(AnnotationDescription.Builder.ofType(Component.class)
                .build())

            .defineMethod("consume", void.class, Modifier.PUBLIC)
            .withParameter(MyPayload.class, "myPayload")
            .annotateParameter(AnnotationDescription.Builder.ofType(Payload.class)
                .build())
            .intercept(MethodDelegation.to(DynamicKafkaConsumerIntercept.class))

            .visit(new MemberAttributeExtension.ForMethod()
                .annotateMethod(AnnotationDescription.Builder.ofType(KafkaListener.class)
                    .defineArray("topics", new String[]{"my-topic"})
                    .define("groupId", "my-groupId")
                    .build())
                .on(named("consume")))

            .make()
            .load(getClass().getClassLoader())
            .getLoaded();
    }

    // simple interceptor for DynamicKafkaConsumer
    public static class DynamicKafkaConsumerIntercept {

        public static void consume(@Argument(0) MyPayload myPayload) {
            System.out.println("Received myPayload: " + myPayload);
        }

    }

}

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;

@Configuration("")
@EnableKafka
public class DynamicKafkaListenerConfig {
    public DynamicKafkaListenerConfig() {
    }
}

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

import com.example.MyPayload;
import io.tpd.kafkaexample.MyBeanFactoryAware.DynamicKafkaConsumerIntercept;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component("")
public class DynamicKafkaListener {
    @KafkaListener(
        concurrency = "",
        topics = {"my-topic"},
        topicPartitions = {},
        errorHandler = "",
        autoStartup = "",
        topicPattern = "",
        id = "",
        groupId = "my-groupId",
        beanRef = "__listener",
        clientIdPrefix = "",
        containerGroup = "",
        containerFactory = "",
        idIsGroup = true
    )
    public void consume(@Payload(required = true,expression = "",value = "") MyPayload myPayload) {
        DynamicKafkaConsumerIntercept.consume(var1);
    }

    public DynamicKafkaListener() {
    }
}

import com.fasterxml.jackson.annotation.JsonProperty;

public class MyPayload {

    private final String message;
    private final int identifier;

    public MyPayload(@JsonProperty("message") final String message,
                     @JsonProperty("identifier") final int identifier) {
        this.message = message;
        this.identifier = identifier;
    }

    public String getMessage() {
        return message;
    }

    public int getIdentifier() {
        return identifier;
    }

    @Override
    public String toString() {
        return "MyPayload{" +
            "message='" + message + '\'' +
            ", identifier=" + identifier +
            '}';
    }

}

  public void consume(@Payload MyPayload myPayload){
      ...
  }

  public void consume(Message<?> message){
     Object payload = message.getPayload();
     MessageHeaders headers = message.getHeaders();
     ...
  }

        DataSourceBuilder dataSourceBuilder = DataSourceBuilder.create();
        dataSourceBuilder.driverClassName("org.postgresql.Driver");
        dataSourceBuilder.url("jdbc:postgresql://127.0.0.1:5432/myDb");
        dataSourceBuilder.username("myUser");
        dataSourceBuilder.password("MyPassword");
        DataSource dataSource = dataSourceBuilder.build();

        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        List<Map<String, Object>> maps = jdbcTemplate.queryForList("select * from my_table");
        _... do something for your case..._

       Environment environment = beanFactory.getBean(Environment.class);
       String url = environment.getProperty("db-url");
       String username = environment.getProperty("db-username");
       String password = environment.getProperty("db-password");
...