假设有两个kafka服务器，一个用于spring云数据流，另一个用于数据源。我创建了一个基于springboot的应用程序，并使用kafka依赖关系和 @KafkaListener 作为源并输出消息。但是，当我将jar文件部署到spring云数据流服务器并部署流时，消息被发送到kafka，该消息用于数据源，而不是带有spring云数据流的kafka。示例代码给出

@EnableBinding(Source.class)
public class TestSource {

    @Autowired
    private Source source;

    @KafkaListener(topics = "topic.datasource")
    public void source(ConsumerRecord<String, GenericRecord> record) throws InterruptedException {

        logger.info("recv {} : {}", record.offset(), record.value());
        // some data pre processing
        source.output().send(MessageBuilder.withPayload(message).build());
    }

我的问题是如何从一个kafka服务器获取数据，并使用spring云数据流将消息发送到另一个kafka服务器。