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

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

使用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,它不是空的,所以它不应该抛出任何这样的错误...

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

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

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