Comment utiliser Twitter Heron avec Storm Flux

Question

Je suis en train de migrer un projet de apache-storm à twitter-heron. Après beaucoup de lutte, j'ai été en mesure de se débarrasser de la plupart des erreurs, telles que className: "org.apache.storm.kafka.ZkHosts" utilisation au lieu de className: "storm.kafka.ZkHosts". Cependant, je suis coincé dans la soumission de la topologie. J'utilise Flux pour soumettre la topologie à l'orage.

Je reçois une exception NullPointerException lors de la création d'un objet CuratorFramework dans ZkState. En creusant plus loin j'ai trouvé an issue in github où il indique, ce problème est provoqué si les configurations au sujet du zookeeper n'ont pas été établies. En outre, j'ai trouvé le problème parce qu'il me manque les configurations suivantes qui sont requises dans ZkState.java:46.

storm.zookeeper.session.timeout 
storm.zookeeper.connection.timeout 
storm.zookeeper.retry.times 
storm.zookeeper.retry.interval 

Pendant que je suis parvenu à identifier le problème, mais je ne suis pas sûr où ajouter dans ma config. Quelqu'un peut-il s'il vous plaît m'aider dans où ajouter la configuration ci-dessus. Je vous remercie.

Mon flux Config

name: "My_Topology" 
components: 
    - id: "zkHosts" 
    className: "org.apache.storm.kafka.ZkHosts" 
    constructorArgs: 
     - "localhost:2181" 

    - id: "SpoutConfig" 
    className: "org.apache.storm.kafka.SpoutConfig" 
    constructorArgs: 
     - ref: "zkHosts" # brokerHosts 
     - "my-topic" # topic 
     - "/my-zkRoot" # zkRoot 
     - "my-id" # spoutId 
    properties: 
     - name: "zkServers" 
     value: ["localhost"] 
     - name: "zkPort" 
     value: 2181 
     - name: "zkRoot" 
     value: "/my-zkRoot" 
     - name: "retryInitialDelayMs" 
     value: 2000 
     - name: "retryDelayMultiplier" 
     value: 2 

config: 
    topology.workers: 5 
    topology.testing.always.try.serialize: true 

spouts: 
    - id: "kafka-spout" 
    className: "org.apache.storm.kafka.KafkaSpout" 
    parallelism: 1 
    constructorArgs: 
     - ref: "SpoutConfig" 
bolts: 
    - id: "my-bolt" 
    className: "com.example.sample.MyBolt" 
    parallelism: 1 

streams: 
    - name: "kafka_spout --> my_bolt" 
    from: "kafka-spout" 
    to: "my-bolt" 
    grouping: 
     type: SHUFFLE 

Answer 1

vous pouvez les ajouter à la section de configuration à votre fichier Flux yaml

config: 
    topology.workers: 5 
    topology.testing.always.try.serialize: true 
    storm.zookeeper.session.timeout: 30000 
    storm.zookeeper.connection.timeout: 30000 
    storm.zookeeper.retry.times: 5 
    storm.zookeeper.retry.interval: 2000