来自Netcat源的事件不通过Kafka通道

Question

我使用水槽代理通过水槽代理收集外部数据。外部数据批次几乎是每 10 秒 1MB。我按如下方式配置了水槽代理。

# Flume agent configuration as /flume/conf/agent.conf
agent.sources = netcat-source
agent.channels = kafka-channel
agent.sinks = logger-sink

########################################
#   Netcat Source
########################################

agent.sources.netcat-source.type = netcat
agent.sources.netcat-source.bind = 0.0.0.0
agent.sources.netcat-source.port = 4141
agent.sources.netcat-source.max-line-length = 500000
agent.sources.netcat-source.channels = kafka-channel

########################################
#   Kafka Channel
########################################

agent.channels.kafka-channel.type =  org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafka-channel.brokerList = 10.212.136.108:9092,10.212.136.108:9092
agent.channels.kafka-channel.zookeeperConnect = 10.212.136.108:2181,10.212.136.108:2181/kafka
agent.channels.kafka-channel.topic = channel
agent.channels.kafka-channel.groupId = fcd-group


########################################
#   Logger Sink
########################################

agent.sinks.logger-sink.type = logger
agent.sinks.logger-sink.channel = kafka-channel

我按以下方式激活了代理。

flume-ng agent -n agent -c /flume/conf -f /flume/conf/agent.conf 

可惜后来发现netcat source运行良好，channel或者sink出了问题。从Ubuntu的资源监视器，我可以看到以下性能。网络性能。蓝色曲线表示输入，而红色曲线表示在没有其他应用程序运行网络io的情况下的输出，我确信这个图展示了我的Flume代理发生了什么。

当我通过控制台消费者查看主题“频道”中的Kafka内容时，我一无所获。还有，当我检查水槽时。日志，我只得到了Flume的状态输出，没有数据。

我已使用 验证了传入的数据

nc -lk 4141 >> my_data_check_file

我的渠道或水槽有什么问题？

另外，当我使用内存通道、文件通道时，事情也变得同样棘手。

Answer 1

啊，终于我自己解决了这个问题！

关键点是行分隔符“\n”。

在Flume源代码NetcatSource中。java，我们有如下一行

private int processEvents(CharBuffer buffer, Writer writer) throws IOException {
  int numProcessed = 0;

  boolean foundNewLine = true;
  while (foundNewLine) {
    foundNewLine = false;

    int limit = buffer.limit();
    for (int pos = buffer.position(); pos < limit; pos++) {
      if (buffer.get(pos) == '\n') {  
        // parse event body bytes out of CharBuffer
        buffer.limit(pos); // temporary limit
        ByteBuffer bytes = Charsets.UTF_8.encode(buffer);
        buffer.limit(limit); // restore limit
... ...
... ...

代码强制输入数据以“\n”结尾。否则，频道不会占用任何事件。我们可以根据需要更改此字符，并将自定义源代码放入 $FLUME_HOME/lib 中