将Storm 1 Kafka拓扑转换为Heron，有几个问题

Question

一直在尝试将Storm 1.0.6拓扑切换到Heron。迈出一小步，去掉卡夫卡的喷口，看看事情进展如何。主要方法如下(从最初的Flux版本修改而来(：

import org.apache.heron.eco.Eco;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class KafkaTopology {
public static void main(String[] args) throws Exception {
List<String> argList = new ArrayList<String>(Arrays.asList(args));
String file = KafkaTopology.class.getClassLoader().getResource("topology.yaml").getFile();
argList.add("local");
argList.add("--eco-config-file");
argList.add(file);
file = KafkaTopology.class.getClassLoader().getResource("dev.properties").getFile();
argList.add("--props");
argList.add(file);
argList.add("--sleep");
argList.add("36000000");
String[] ecoArgs = argList.toArray(new String[argList.size()]);
Eco.main(ecoArgs);
}
}

YAML是这样的吗：

name: "kafkaTopology-XXX_topologyVersion_XXX"
type: "storm"
config:
topology.workers: ${workers.config}
topology.max.spout.pending: ${max.spout.pending}
topology.message.timeout.secs: 120
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
properties:
kafka.mapper.zkServers: ${kafka.mapper.zkServers}
kafka.mapper.zkPort: ${kafka.mapper.zkPort}
bootstrap.servers: ${bootstrap.servers}
kafka.mapper.brokerZkStr: ${kafka.mapper.brokerZkStr}
kafka.topic.name: ${kafka.topic.name}
components:
- id: "zkHosts"
className: "org.apache.storm.kafka.ZkHosts"
constructorArgs:
- ${kafka.mapper.brokerZkStr}
- id: "rawMessageAndMetadataScheme"
className: "org.acme.storm.spout.RawMessageAndMetadataScheme"
- id: "messageMetadataSchemeAsMultiScheme"
className: "org.apache.storm.kafka.MessageMetadataSchemeAsMultiScheme"
constructorArgs:
- ref: "rawMessageAndMetadataScheme"
- id: "kafkaSpoutConfig"
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
# brokerHosts
- ref: "zkHosts"
# topic
- ${kafka.topic.name}
# zkRoot
- "/zkRootKafka.kafkaSpout.builder"
# id
- ${kafka.topic.name}
properties:
- name: "scheme"
ref: "messageMetadataSchemeAsMultiScheme"
- name: zkServers
value: ${kafka.mapper.zkServers}
- name: zkPort
value: ${kafka.mapper.zkPort}
# Retry Properties
- name: "retryInitialDelayMs"
value: 60000
- name: "retryDelayMultiplier"
value: 1.5
- name: "retryDelayMaxMs"
value: 14400000
- name: "retryLimit"
value: 0
# spout definitions
spouts:
- id: "kafka-spout"
className: "org.apache.storm.kafka.KafkaSpout"
parallelism: ${kafka.spout.parallelism}
constructorArgs:
- ref: "kafkaSpoutConfig"

相关POM条目：

<dependency>
<groupId>org.apache.heron</groupId>
<artifactId>heron-api</artifactId>
<version>0.20.3-incubating</version>
</dependency>
<dependency>
<groupId>org.apache.heron</groupId>
<artifactId>heron-storm</artifactId>
<version>0.20.3-incubating</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>1.0.6</version>
</dependency>

主要方法似乎运行良好：

Apr 30, 2021 4:38:49 PM org.apache.heron.eco.parser.EcoParser loadTopologyFromYaml
INFO: Parsing eco config file
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.parser.EcoParser loadTopologyFromYaml
INFO: Performing property substitution.
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.parser.EcoParser loadTopologyFromYaml
INFO: Performing environment variable substitution.
topology type is Storm
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.builder.storm.EcoBuilder buildConfig
INFO: Building topology config
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: ---------- TOPOLOGY DETAILS ----------
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: Topology Name: kafkaTopology-XXX_topologyVersion_XXX
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: --------------- SPOUTS ---------------
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: kafka-spout [1] (org.apache.storm.kafka.KafkaSpout)
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: ---------------- BOLTS ---------------
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: --------------- STREAMS ---------------
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.Eco printTopologyInfo
INFO: --------------------------------------
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.builder.storm.EcoBuilder buildTopologyBuilder
INFO: Building components
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.builder.storm.EcoBuilder buildTopologyBuilder
INFO: Building spouts
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.builder.storm.EcoBuilder buildTopologyBuilder
INFO: Building bolts
Apr 30, 2021 4:38:49 PM org.apache.heron.eco.builder.storm.EcoBuilder buildTopologyBuilder
INFO: Building streams
Process finished with exit code 0

问题1:拓扑立即退出，是否有等效于Flux"--sleep"的Eco标志使其运行一段时间(调试等(？

问题2:我有点惊讶于我需要引入storm kafka(认为会有一个Heron等价物(-这是正确的吗(或者其他工件？(如果是，1.0.6是一个可以使用的版本吗？或者Heron与另一个版本配合得更好吗？

问题3：以上是YAML中的type: "storm"，尝试type: "heron"会出现以下错误：

INFO: Building spouts
Exception in thread "main" java.lang.ClassCastException: class org.apache.storm.kafka.KafkaSpout cannot be cast to class org.apache.heron.api.spout.IRichSpout (org.apache.storm.kafka.KafkaSpout and org.apache.heron.api.spout.IRichSpout are in unnamed module of loader 'app')
at org.apache.heron.eco.builder.heron.SpoutBuilder.buildSpouts(SpoutBuilder.java:42)
at org.apache.heron.eco.builder.heron.EcoBuilder.buildTopologyBuilder(EcoBuilder.java:70)
at org.apache.heron.eco.Eco.submit(Eco.java:125)
at org.apache.heron.eco.Eco.main(Eco.java:161)
at KafkaTopology.main(KafkaTopology.java:26)
Process finished with exit code 1

这只是使用卡夫卡的方式吗，类型需要是风暴而不是苍鹭，或者这里有一些变通方法吗？

Answer 1

Question 1：我不确定拓扑结构为什么会在您身上关闭。尝试使用--verbose标志运行提交。此时，--sleep参数的功能不存在。如果你需要的话，可以添加一个功能。

Question 2：有一个Heron等价物。在Heron被捐赠给Apache之后，必须做大量的工作才能发布二进制版本。大部分工作已经完成。在下一个版本中，我希望所有二进制工件都能得到适当的分发。

Question 3：出现此问题是因为它根据指定的类型在某个包中查找螺栓/喷嘴。当"；风暴；它期望它实现或扩展的类为"0"；org.apache.storm"；。当"；苍鹭；它期望它实现或扩展的类为"0"；org.apache.heron"；。如果您使用依赖风暴kafka，则类型将需要是"；风暴"；。鹭的等价物可以在这里找到。https://search.maven.org/search?q=heron-卡夫卡

Answer 2

2. https://search.maven.org/search?q=heron-卡夫卡

• Heron有几个卡夫卡的口头禅。我使用Storm(Storm-kafclient-2.1(克隆并在生产中使用它
• https://search.maven.org/artifact/com.github.thinker0.heron/heron-kafka-client/1.0.4.1/jar