Apache Flume Java 客户端无法使用单个 Kafka 接收器启动

Question

使用org.apache.flume.agent.embedded.EmbeddedAgent。配置如下：

Map<String, String> configurationProperties = ...;
service.configure(configurationProperties);

其中 configurationProperties 设置为：

    {
      "kafkaSink.kafka.producer.reconnect.backoff.max.ms": "30000",
      "processor.type": "load_balance",
      "sinks": "kafkaSink1",
      "channel.keep-alive": "0",
      "channel.checkpointDir": "********************************",
      "kafkaSink.kafka.producer.reconnect.backoff.ms": "2000",
      "channel.dataDirs": "********************************",
      "kafkaSink.kafka.producer.retry.backoff.ms": "1000",
      "processor.selector.maxTimeOut": "60000",
      "kafkaSink.kafka.producer.max.request.size": "5485760",
      "kafkaSink1.flumeBatchSize": "1000",
      "kafkaSink.kafka.producer.buffer.memory": "67108864",
      "kafkaSink.kafka.producer.client.id": "********************************",
      "kafkaSink1.useFlumeEventFormat": "true",
      "kafkaSink1.kafka.topic": "********************************",
      "kafkaSink.kafka.producer.batch.size": "8196",
      "channel.kafka.dataDirs": "********************************",
      "kafkaSink1.type": "KAFKA",
      "channel.backupCheckpointDir": "********************************",
      "kafkaSink1.allowTopicOverride": "true",
      "channel.useDualCheckpoints": "true",
      "kafkaSink.kafka.producer.compression.type": "lz4",
      "processor.maxBackoff": "60000",
      "use_dual_channel": "true",
      "channel.capacity": "1000000",
      "channel.byteCapacityBufferPercentage": "50",
      "channel.transactionCapacity": "1000",
      "channel.byteCapacity": "10485760",
      "channel.type": "file",
      "processor.backoff": "true",
      "channel.kafka.checkpointDir": "********************************",
      "channel.kafka.backupCheckpointDir": "********************************",
      "kafkaSink1.kafka.bootstrap.servers": "********************************",
      "kafkaSink.kafka.producer.acks": "-1"
    }

在运行时它会抛出以下内容：

java.lang.NullPointerException
        at org.apache.flume.conf.sink.SinkGroupConfiguration.configure(SinkGroupConfiguration.java:52)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateGroups(FlumeConfiguration.java:927)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:384)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access[=14=]0(FlumeConfiguration.java:228)
        at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:153)
        at org.apache.flume.conf.FlumeConfiguration.<init>(FlumeConfiguration.java:133)
        at org.apache.flume.agent.embedded.MemoryConfigurationProvider.getFlumeConfiguration(MemoryConfigurationProvider.java:45)
        at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:97)
        at org.apache.flume.agent.embedded.MaterializedConfigurationProvider.get(MaterializedConfigurationProvider.java:40)
        at org.apache.flume.agent.embedded.EmbeddedAgent.doConfigure(EmbeddedAgent.java:161)
        at org.apache.flume.agent.embedded.EmbeddedAgent.configure(EmbeddedAgent.java:99)
        at ******************.startService(******************)

并且flume将不会启动。

org.apache.flume.conf.sink.SinkGroupConfiguration.configure(SinkGroupConfiguration.java:52 的代码显示它正在寻找属性 sinks，它不是空的，所以它不应该抛出任何这样的错误...

有人知道为什么吗？没有关于这些的任何文件....

Answer 1

万一有人偶然发现这个问题... 在进一步调查之后，我相信 Apache Flume 项目已经死了，我们将停止使用它。详情请看以下内容：

https://medium.com/@bmedalen/great-post-again-prathamesh-i-really-enjoy-your-content-posts-cc0b831a3811
许多未解决的问题……没有回应……https://cwiki.apache.org/confluence/display/FLUME/Developer+Section
最后一个版本大约是 2 年前发布的（最初它每 6 个月发布一次）。 https://flume.apache.org/index.html
Wiki 甚至没有更新最新版本 - 最后一次更新是在 5 年前 - https://cwiki.apache.org/confluence/display/FLUME/Home

遗憾的是，我将不得不以这种方式结束这个问题。希望对大家有帮助。

Apache Flume Java 客户端无法使用单个 Kafka 接收器启动

Apache Flume Java client fails to start with a single Kafka sink

java

apache-kafka

flume

flume-ng