KafkaSource 配置topic：topic1
KafkaSink 配置topic：topic2
从topic1拉取数据，经过简单处理后发到topic2
但是你会发现flume一直在循环读写topic1
原因就是KafkaSink中的这段代码：

1
2
3

if (eventTopic == null) {
    eventTopic = topic;
} 

首先从headers中获取TOPIC_HEADER(topic),然后优先使用。然而在KafkaSource中则会将topic1 PUT到Header中，所以导致循环读写topic1。

如何解决呢？

你可以重新KafkaSink，改掉上面那段代码。

或者配置一个拦截器，将KafkaSource 写到Header中的topic给替换掉：

1
2
3
4

agent.sources.so1.interceptors.i1.type = static
agent.sources.so1.interceptors.i1.key = topic
agent.sources.so1.interceptors.i1.preserveExisting = false
agent.sources.so1.interceptors.i1.value = topic2

千万别用高版本的kafka client 去连低版本的kafka server。
否则回报一系列的(n)io异常，比如下面这个异常

  java.nio.BufferUnderflowException

不丢消息：producer有个ack参数，有三个值，分别代表：不在乎是否写入成功、写入leader成功、写入leader和所有reclpica成功；要求非常可靠的话可以牺牲性能设置成最后一种。
不重复发送：正常发都不会重复，只可能丢，看你这边怎么容错重发了，参考上一条。
消息只读一次：同样，正常读不会重复，如果在上一次读的过程中发生了异常，消息可能被消费，但是offset没有及时commit；这本身是两步，存在中间crash的风险