【发布时间】：2019-02-19 21:26:01
【问题描述】：
我尝试做的场景如下：
1- Flume TAILDIR 源从日志文件中读取并将静态拦截器附加到消息的开头。拦截器由主机名和主机 IP 组成，因为我收到的每条日志消息都需要它。
2- Flume Kafka Producer Sink 从文件中获取这些消息并将它们放入 Kafka 主题中。
Flume配置如下：
tier1.sources=source1
tier1.channels=channel1
tier1.sinks =sink1
tier1.sources.source1.interceptors=i1
tier1.sources.source1.interceptors.i1.type=static
tier1.sources.source1.interceptors.i1.key=HostData
tier1.sources.source1.interceptors.i1.value=###HostName###000.00.0.000###
tier1.sources.source1.type=TAILDIR
tier1.sources.source1.positionFile=/usr/software/flumData/flumeStressAndKafkaFailureTestPos.json
tier1.sources.source1.filegroups=f1
tier1.sources.source1.filegroups.f1=/usr/software/flumData/flumeStressAndKafkaFailureTest.txt
tier1.sources.source1.channels=channel1
tier1.channels.channel1.type=file
tier1.channels.channel1.checkpointDir = /usr/software/flumData/checkpoint
tier1.channels.channel1.dataDirs = /usr/software/flumData/data
tier1.sinks.sink1.channel=channel1
tier1.sinks.sink1.type=org.apache.flume.sink.kafka.KafkaSink
tier1.sinks.sink1.kafka.bootstrap.servers=<Removed For Confidentiality >
tier1.sinks.sink1.kafka.topic=FlumeTokafkaTest
tier1.sinks.sink1.kafka.flumeBatchSize=20
tier1.sinks.sink1.kafka.producer.acks=0
tier1.sinks.sink1.useFlumeEventFormat=true
tier1.sinks.sink1.kafka.producer.linger.ms=1
tier1.sinks.sink1.kafka.producer.client.id=HOSTNAME
tier1.sinks.sink1.kafka.producer.compression.type = snappy
所以现在我正在测试，我运行了一个控制台 Kafka 消费者并开始在源文件中写入，我确实收到了带有标题的消息。
例子：
我在源文件中写入'test'并回车然后保存文件
Flume 检测到文件更改，然后将新行发送给 Kafka 生产者。
我的消费者得到以下行：
###HostName###000.00.0.000###test
现在的问题是，有时拦截器无法按预期工作。就像 Flume 发送 2 条消息，一条包含拦截器，另一条包含消息内容。
例子：
我在源文件中写上'hi you'然后回车然后保存文件
Flume 检测到文件更改，然后将新行发送给 Kafka 生产者。
我的消费者得到以下 2 行：
###HostName###000.00.0.000###
hi you
终端滚动到新的消息内容。
这种情况总是在我在文本文件中输入“hi you”时发生，而且由于我是从日志文件中读取的，所以它何时发生是不可预测的。
我们将不胜感激帮助和支持^^
谢谢
【问题讨论】：
标签： apache-kafka interceptor flume