java - 如何停止 Flume 的 HTTP 源

Question

我正在使用 Flume 1.4.0，我正在尝试以某种方式停止 Flume 的组件:

• 首先，停止来源。
• 然后等待 channel 内的所有事件都被接收器消耗。
• 消耗完所有事件后，停止 channel 和接收器。

上述任务由一个关闭 Hook 执行，就像在 org.apache.flume.node.Application 中创建的那样(事实上，我正在开发一个自定义的 Application).

我获取对源、 channel 和接收器的引用的方式是:

MaterializedConfiguration conf = configurationProvider.getConfiguration();
ImmutableMap<String, SourceRunner> sourcesRef = conf.getSourceRunners();
ImmutableMap<String, Channel> channelsRef = conf.getChannels();
ImmutableMap<String, SinkRunner> sinksRef = conf.getSinkRunners();

关键是我得到了这个NullPointerException:

2015-02-17 16:03:28,094 (agent-shutdown-hook) [ERROR - org.apache.flume.source.http.HTTPSource.stop(HTTPSource.java:169)] Error while stopping HTTPSource. Exception follows.
java.lang.NullPointerException
    at org.apache.flume.source.http.HTTPSource.stop(HTTPSource.java:165)
    at     org.apache.flume.source.EventDrivenSourceRunner.stop(EventDrivenSourceRunner.java:51)
    at     es.tid.fiware.fiwareconnectors.cygnus.nodes.CygnusApplication$AgentShutdownHook.stopSources(CygnusApplication.java:296)
    at     es.tid.fiware.fiwareconnectors.cygnus.nodes.CygnusApplication$AgentShutdownHook.run(CygnusApplication.java:231)

HTTPSource.java:165 是关于停止实现源的 Http 服务器部分的 Jetty 服务器，它似乎是空的:

162  @Override
163  public void stop() {
164    try {
165      srv.stop();
166      srv.join();
167      srv = null;
168    } catch (Exception ex) {
169      LOG.error("Error while stopping HTTPSource. Exception follows.", ex);
170    }
171    sourceCounter.stop();
172    LOG.info("Http source {} stopped. Metrics: {}", getName(), sourceCounter);
173  }

为什么为空？源工作正常，能够接收 Http 请求。

我猜这不是关闭 Flume 组件的正确方法...如果不是，那是什么？

谢谢!

Answer 1

原因是 srv 被多个线程共享(因此它是 volatile 声明)。 Flume 尝试调用 close 来终止源，并且这种情况发生了不止一次。第二次调用 stop() 失败，因为 srv 已经无效。

这种情况发生在您的情况而不是标准的 Vanilla 水槽代理中的原因可能是因为您没有更新 SourceCounter。查看 MonitoredCounterGroup 了解详细信息。