我正在把Flume里的数据传给Kafka。我的flafka配置文件如下所示：

tier1.sources  = source1
 tier1.channels = channel1
 tier1.sinks    = sink1

 tier1.sources.source1.type = exec
 tier1.sources.source1.command = cat /testing.txt
 tier1.sources.source1.channels = channel1

 tier1.channels.channel1.type   = memory
 tier1.channels.channel1.capacity = 10000
 tier1.channels.channel1.transactionCapacity = 1000

 tier1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
 tier1.sinks.sink1.topic = sink1
 tier1.sinks.sink1.brokerList = kafkagames-1:9092,kafkagames-2:9092
 tier1.sinks.sink1.channel = channel1
 tier1.sinks.sink1.batchSize = 20

我用kafka河插件连接了kafka和elasticsearch。flume source正在向kafka消费者发送数据，而elasticsearch正在从kafka生产者处接收数据。
有没有一种方法可以将数据从flume推送到kafka生产商，而不是直接发送给kafka消费者，以便elasticsearch可以读取数据？
有什么建议吗？谢谢。