java - 从 ElasticsearchIO 等待 [10000] 毫秒后获取监听器超时

Question

我正在尝试测试一个简单的 Apache Beam 代码，其源代码为 Elasticsearch。我从git repo找到了ElasticsearchIO源类。

我修改了 Beam 的 MinimalWordCount 示例，将源包含为 Elasticsearch 而不是 TextIO。以下是要点，

String[] hosts = new String[1];
hosts[0]="http://localhost:9200";
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
p.apply(
      ElasticsearchIO.read().withConnectionConfiguration(
        ElasticsearchIO.ConnectionConfiguration.create(hosts, "test_index", "users").withUsername("esuser").withPassword("password")
      )
 )
.apply("ExtractWords", ParDo.of(new DoFn<String, String>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
         for (String word : c.element().split("[^a-zA-Z']+")) { 
           if (!word.isEmpty()) {
             c.output(word);
           }
         }
      }
}));
p.run().waitUntilFinish();

如果我运行代码，

mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.MinimalWordCount \
 -Pdirect-runner

我收到错误

An exception occured while executing the Java class. null: InvocationTargetException: java.io.IOException: listener timeout after waiting for [10000] ms

我调试了ElasticsearchIO.java可以看到一切正常，Elasicsearch 客户端已构建并且代码正在检索索引中的数据。但是读取转换后的 ParDo 函数根本没有被执行。 Elasticsearch 客户端一直等待，最后出现超时错误。

我知道 Beam 的 Elasicsearch 连接器仍在开发中。但是有人可以帮我找出我做错了什么吗？

PS:我在本地运行 Elasticsearch 5.2.1。

Answer 1

当前版本的 ElasticsearchIO (beam-0.6.0 2017/03) 尚不支持 Elasticsearch 版本 5。这方面的进展由 https://issues.apache.org/jira/browse/BEAM-1637 跟踪。