flume+kafka，kafka日志存储暴增，数据重复传输

以下是错误配置

agent.sources = kafkaSource1
agent.channels = kafkaChannel
agent.sinks = sqlSink
agent.sources.kafkaSource1.channels = kafkaChannel
agent.sinks.sqlSink.channel = kafkaChannel

######source config

agent.sources.kafkaSource1.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafkaSource1.zookeeperConnect = *.*.*.2:2181,*.*.*.3:2181,*.*.*.4:2181
agent.sources.kafkaSource1.topic = test
agent.sources.kafkaSource1.groupId = sqlgroup
agent.sources.kafkaSource1.kafka.consumer.timeout.ms = 100
agent.sources.kafkaSource1.kafka.bootstrap.servers = *.*.*.2:6667,*.*.*.3:6667,*.*.*.5:6667
agent.sources.kafkaSource1.batchSize = 1000
agent.sources.kafkaSource1.batchDurationMillis = 1000

######channel config

agent.channels.kafkaChannel.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafkaChannel.kafka.bootstrap.servers = *.*.*.2:6667,*.*.*.3:6667,*.*.*.5:6667
agent.channels.kafkaChannel.kafka.topic = test
agent.channels.kafkaChannel.kafka.consumer.group.id = sqlgroup-consumer
agent.channels.kafkaChannel.parseAsFlumeEvent = false

######sink config
#---------Sink 相关配置------------------

kafka source用consumer来从kafka pull数据，kafka sink使用producer将数据发到kafka。而kafka channel包含producer和consumer，producer接受source发送过来的数据放到broker中，consumer从broker中pull数据给sink。目前只允许有一个topic，而且kafka自己的参数可以加上"kafka.*"添加到flume的配置文件。
因此，kafka source用consumer从kafka pull数据，又在kafka channel中将数据重复再次写入kafka,从而导致数据在kafka中无限循环。