3

我正在使用 Storm Flux (0.10.0) DSL 部署以下拓扑（简化为仅保留相关部分）：

---
name: "my-topology"

components:
  - id: "esConfig"
    className: "java.util.HashMap"
    configMethods:
      - name: "put"
        args:
          - "es.nodes"
          - "${es.nodes}"

bolts:
  - id: "es-bolt"
    className: "org.elasticsearch.storm.EsBolt"
    constructorArgs:
      - "myindex/docs"
      - ref: "esConfig"
    parallelism: 1

# ... other bolts, spouts and streams here

如您所见，我使用的org.elasticsearch.storm.EsBolt其中一个螺栓具有以下构造函数（参见代码）：

public EsBolt(String target) { ... }
public EsBolt(String target, boolean writeAck) { ... }
public EsBolt(String target, Map configuration) { ... }

应该调用最后一个，因为我在 constructorArgs 中传递了一个 String 和一个 Map。但是当我部署拓扑时，出现以下异常，好像 Flux 无法从类型（字符串、映射）中推断出正确的构造函数：

storm jar mytopology-1.0.0.jar org.apache.storm.flux.Flux --local mytopology.yml --filter mytopology.properties

...
Version: 0.10.0
Parsing file: mytopology.yml
958  [main] INFO  o.a.s.f.p.FluxParser - loading YAML from input stream...
965  [main] INFO  o.a.s.f.p.FluxParser - Performing property substitution.
969  [main] INFO  o.a.s.f.p.FluxParser - Not performing environment variable substitution.
1252 [main] INFO  o.a.s.f.FluxBuilder - Detected DSL topology...
1431 [main] WARN  o.a.s.f.FluxBuilder - Found multiple invokable constructors for class class org.elasticsearch.storm.EsBolt, given arguments [myindex/docs, {es.nodes=localhost}]. Using the last one found.
Exception in thread "main" java.lang.IllegalArgumentException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
    at org.apache.storm.flux.FluxBuilder.buildObject(FluxBuilder.java:291)
    at org.apache.storm.flux.FluxBuilder.buildBolts(FluxBuilder.java:372)
    at org.apache.storm.flux.FluxBuilder.buildTopology(FluxBuilder.java:88)
    at org.apache.storm.flux.Flux.runCli(Flux.java:153)
    at org.apache.storm.flux.Flux.main(Flux.java:98)

对可能发生的事情有任何想法吗？以下是Storm Flux 如何找到兼容的构造函数。神奇在于canInvokeWithArgs方法。

这些是 Flux 调试日志，您可以在其中看到 FluxBuilder 如何找到最合适的构造函数：

Version: 0.10.0
Parsing file: mytopology.yml
559  [main] INFO  o.a.s.f.p.FluxParser - loading YAML from input stream...
566  [main] INFO  o.a.s.f.p.FluxParser - Performing property substitution.
569  [main] INFO  o.a.s.f.p.FluxParser - Not performing environment variable substitution.
804  [main] INFO  o.a.s.f.FluxBuilder - Detected DSL topology...
org.apache.logging.slf4j.Log4jLogger@3b69e7d1
1006 [main] DEBUG o.a.s.f.FluxBuilder - Found constructor arguments in definition: java.util.ArrayList
1006 [main] DEBUG o.a.s.f.FluxBuilder - Checking arguments for references.
1010 [main] DEBUG o.a.s.f.FluxBuilder - Target class: org.elasticsearch.storm.EsBolt
1011 [main] DEBUG o.a.s.f.FluxBuilder - found constructor with same number of args..
1011 [main] DEBUG o.a.s.f.FluxBuilder - Comparing parameter class class java.lang.String to object class class java.lang.String to see if assignment is possible.
1011 [main] DEBUG o.a.s.f.FluxBuilder - Yes, they are the same class.
1012 [main] DEBUG o.a.s.f.FluxBuilder - ** invokable --> true
1012 [main] DEBUG o.a.s.f.FluxBuilder - found constructor with same number of args..
1012 [main] DEBUG o.a.s.f.FluxBuilder - Comparing parameter class class java.lang.String to object class class java.lang.String to see if assignment is possible.
1012 [main] DEBUG o.a.s.f.FluxBuilder - Yes, they are the same class.
1012 [main] DEBUG o.a.s.f.FluxBuilder - ** invokable --> true
1012 [main] DEBUG o.a.s.f.FluxBuilder - Skipping constructor with wrong number of arguments.
1012 [main] WARN  o.a.s.f.FluxBuilder - Found multiple invokable constructors for class class org.elasticsearch.storm.EsBolt, given arguments [myindex/docs, {es.nodes=localhost}]. Using the last one found.
1014 [main] DEBUG o.a.s.f.FluxBuilder - Found something seemingly compatible, attempting invocation...
1044 [main] DEBUG o.a.s.f.FluxBuilder - Comparing parameter class class java.lang.String to object class class java.lang.String to see if assignment is possible.
1044 [main] DEBUG o.a.s.f.FluxBuilder - They are the same class.
1044 [main] DEBUG o.a.s.f.FluxBuilder - Comparing parameter class boolean to object class class java.util.HashMap to see if assignment is possible.
Exception in thread "main" java.lang.IllegalArgumentException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
    at org.apache.storm.flux.FluxBuilder.buildObject(FluxBuilder.java:291)
...
4

2 回答 2

1

此问题已于 2015 年 11 月 18 日修复。

看到这个：https ://github.com/apache/storm/commit/69b9cf581fd977f6c28b3a78a116deddadc44014

带有此修复程序的下一个 Storm 版本应该会在一个月内发布。

于 2016-03-12T04:33:57.997 回答
0

作为一个糟糕的解决方法，我最终扩展EsBolt为只公开我需要的构造函数并避免冲突。

package com.mypackage;

import java.util.Map;
import org.elasticsearch.storm.EsBolt;

public class EsBoltWrapper extends EsBolt {

    public EsBoltWrapper(String target) {
        super(target);
    }

    public EsBoltWrapper(String target, Map configuration) {
        super(target, configuration);
    }
}

现在我的拓扑是这样的：

bolts:
  - id: "es-bolt"
    className: "com.mypackage.EsBoltWrapper" # THE NEW CLASS
    constructorArgs:
      - "myindex/docs"
      - ref: "esConfig"
    parallelism: 1

这似乎是 Storm Flux 中的一个错误。

于 2016-02-24T14:42:07.207 回答