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 项目已经死了,我们将停止使用它。
详情请看以下内容:
-
许多未解决的问题……没有回应……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
遗憾的是,我将不得不以这种方式结束这个问题。
希望对大家有帮助。
使用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 项目已经死了,我们将停止使用它。 详情请看以下内容:
许多未解决的问题……没有回应……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
遗憾的是,我将不得不以这种方式结束这个问题。 希望对大家有帮助。