java - KafkaSpout 没有收到来自 Kafka 的任何信息

Question

我正在尝试装配一个 Kafka-Storm“Hello World”系统。我安装并运行了 Kafka，当我使用 Kafka 生产者发送数据时，我可以使用 Kafka 控制台消费者读取它。

我从 O'Reilly 的书“Getting Started With Storm”中获取了第 02 章的示例，并将其修改为使用 KafkaSpout 而不是常规的 spout。

当我运行应用程序时，数据已经在 kafka 中挂起，KafkaSpout 的 nextTuple 没有收到任何消息 - 它进入，尝试遍历协调器下的空管理器列表，然后退出。

我的环境是一个相当老的 Cloudera VM，有 Storm 0.9 和 Kafka-Storm-0.9(最新的)，以及 Kafka 2.9.2-0.7.0。

这就是我定义 SpoutConfig 和拓扑的方式:

String zookeepers = "localhost:2181";

SpoutConfig spoutConfig = new SpoutConfig(new SpoutConfig.ZkHosts(zookeepers, "/brokers"),
        "gtest",
        "/kafka",  // zookeeper root path for offset storing
        "KafkaSpout");
spoutConfig.forceStartOffsetTime(-1);

KafkaSpoutTester kafkaSpout = new KafkaSpoutTester(spoutConfig);


//Topology definition
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("word-reader", kafkaSpout, 1);
builder.setBolt("word-normalizer", new WordNormalizer())
    .shuffleGrouping("word-reader");
builder.setBolt("word-counter", new WordCounter(),1)
    .fieldsGrouping("word-normalizer", new Fields("word"));

//Configuration
Config conf = new Config();
conf.put("wordsFile", args[0]);
conf.setDebug(false);
//Topology run
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
cluster = new LocalCluster();
cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology());

谁能帮我弄清楚为什么我没有收到任何东西？

谢谢，G.

Answer 1

如果您已经使用了该消息，则不应再读取它，除非您的生产者生成新消息。这是因为 forceStartOffsetTime 在您的代码中使用 -1 调用。

表格storm-contrib文档:

Another very useful config in the spout is the ability to force the spout to rewind to a previous offset. You do forceStartOffsetTime on the spout config, like so:

   spoutConfig.forceStartOffsetTime(-2);

它将选择围绕该时间戳写入的最新偏移量开始消费。您可以通过传入 -1 强制 spout 始终从最新的偏移量开始，也可以通过传入 -2 强制它从最早的偏移量开始。

你的制作人长什么样？有一个片段会很有用。您可以将 -1 替换为 -2，看看您是否收到任何东西，如果您的生产者没问题，那么您应该可以消费。