实战经验 | Flume中同时使用Kafka Source和Kafka Sink的Topic覆盖问题

作者:lxw的大数据田地

By 暴走大数据

场景描述：如果在一个Flume Agent中同时使用Kafka Source和Kafka Sink来处理events，便会遇到Kafka Topic覆盖问题，具体表现为，Kafka Source可以正常从指定的Topic中读取数据，但在Kafka Sink中配置的目标Topic不起作用，数据仍然会被写入到Source中指定的Topic中。

关键词：Flume Kafka

问题发现

如果在一个Flume Agent中同时使用Kafka Source和Kafka Sink来处理events，便会遇到Kafka Topic覆盖问题，具体表现为，Kafka Source可以正常从指定的Topic中读取数据，但在Kafka Sink中配置的目标Topic不起作用，数据仍然会被写入到Source中指定的Topic中。

比如：在Agent中的Kafka Source配置Topic为：

agent_myAgent.sources.kafkaSource.topic = sourceTopic

在Kafka Sink配置Topic为：

agent_myAgent.sinks.kafkaSink.topic = sinkTopic

你会发现，最后数据又被写入到sourceTopic中，而sinkTopic没有任何数据写入。

经过DEBUG和分析，原因如下：在Kafka Sink中，配置项官网文档说明如下：

属性名topic，默认值为default-flume-topic。

The topic in Kafka to which the messages will be published. If this parameter is configured, messages will be published to this topic. If the event header contains a “topic” field, the event will be published to that topic overriding the topic configured here。

如果event header中包含了key为”topic”的值，那么将会覆盖该属性配置。

在源码org.apache.flume.sink.kafka.KafkaSink.process()中，

if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
            eventTopic = topic;
 }

其中topic是从属性agent_myAgent.sinks.kafkaSink.topic = sinkTopic 中获取的属性值（如果没有配置，则使用默认topic名称）

topic = context.getString(KafkaSinkConstants.TOPIC,KafkaSinkConstants.DEFAULT_TOPIC);

即：先使用event header中key为”topic”的值作为sink的topic，如果event header中没有，才取属性中配置的topic。

在Kafka Source中

源码：org.apache.flume.source.kafka.KafkaSource.process()

// Add headers to event (topic, timestamp, and key)headers = new HashMap<String, String>();headers.put(KafkaSourceConstants.TIMESTAMP,String.valueOf(System.currentTimeMillis()));headers.put(KafkaSourceConstants.TOPIC, topic);

在event中，将Kafka Source中配置的topic加入到了header中。

因此，在Kafka Sink中，首先从event header中读取到了topic，Sink端的配置项不起作用。

解决办法

使用Flume拦截器，修改event header中key=topic的值为目标topic，拦截器使用Static interceptor，配置如下：

## Source 拦截器
agent_myAgent.sources.kafkaSource.interceptors = i1
agent_myAgent.sources.kafkaSource.interceptors.i1.type = static
agent_myAgent.sources.kafkaSource.interceptors.i1.key = topic
agent_myAgent.sources.kafkaSource.interceptors.i1.preserveExisting = false
agent_myAgent.sources.kafkaSource.interceptors.i1.value = sinkTopic

其中，要特别注意preserveExisting属性值需要设置为false，该属性设置如果event header中已经存在key=topic，是否保留原来的值，默认为true，如果为true，那么还是会保留原来的topic，你设置的将不会生效。

另外，Kafka Sink中可以不用再设置topic属性了，反正也没用。 agent_myAgent.sinks.kafkaSink.topic = sinkTopic //不需要了