Apache Storm - Kafka Spout 无法连接到 Kafka 主题

Question

我正在尝试连接属于在LocalCluster对象上运行的风暴拓扑的KafkaSpout。我根据我在 https://github.com/apache/storm/tree/master/external/storm-kafka 上找到的文档编写了此代码。

 private static final String brokerZkStr = "localhost:2181";
private static final String topic = "/test-topic-multi";
public void startTopology()
{
    BrokerHosts hosts = new ZkHosts(brokerZkStr);
    SpoutConfig conf = new SpoutConfig(hosts, topic, "localhost:2181", UUID
            .randomUUID().toString());

    KafkaSpout kafkaSput = new KafkaSpout(conf);
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("kafka-spout", kafkaSput);
    Config topConfig = new Config();
    topConfig.setDebug(true);
    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("HelloStorm", topConfig , builder.createTopology());
}

我想使用在localhost:2181运行的 zookeeper 实例，但是当尝试运行代码时，我收到以下错误：

  java.lang.RuntimeException: java.lang.RuntimeException: java.lang.IllegalArgumentException: Invalid path string "/brokers/topics//test-topic-multi/partitions" caused by empty node name specified @16
at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:81)
at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42)
at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57)
at storm.kafka.KafkaSpout.open(KafkaSpout.java:87)

这似乎只是一个设置错误的问题，但我无法解决

PS Kafka 配置如下：1 个 Zookeeper 实例和 2 个在 localhost:9092 上运行的代理和localhost:9093

Answer 1

我想

我解决了它。我只是搞砸了配置代码。正确的是：

private static final String topic = "test-topic-multi";
....
SpoutConfig conf = new SpoutConfig(hosts, topic, "/" + topic, UUID
            .randomUUID().toString());

Answer 2

您的 kafka 主题名称无效。为什么要尝试连接到不存在的主题？

sql@injection:~$ kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic /test-topic
Error while executing topic command topic name /test-topic is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'
kafka.common.InvalidTopicException: topic name /test-topic is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'
    at kafka.common.Topic$.validate(Topic.scala:42)
    at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:181)
    at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:172)
    at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:93)
    at kafka.admin.TopicCommand$.main(TopicCommand.scala:55)
    at kafka.admin.TopicCommand.main(TopicCommand.scala)

您真的确定前斜杠属于主题名称吗？