请问kafka怎么接收MQTT apollo发过来的数据？

我这里的项目是车机端通过MQTT apollo发送到kafka，再将kafka中的数据写到数据库中，但是发送到kafka要怎么去实现？有大神知道思路吗？网上找了一圈没发现有相关的解决方案

@Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setServerURIs("tcp://demo:1883");
//        factory.setUserName("guest");
//        factory.setPassword("guest");
        return factory;
    }

    @Bean
    public IntegrationFlow mqttInFlow() {
        return IntegrationFlows.from(mqttInbound())
                .transform(p -> p + ", received from MQTT")
                .handle(logger())
                .get();
    }