Kafka没有收到Kafka的任何东西

Question

我正试图装配一个KafkaStorm“你好世界”系统。我有Kafka安装和运行，当我发送数据与Kafka生产者我可以读取它与Kafka控制台消费者。

我从O'Reilly的《Storm入门》一书中选取了第二章的例子，并将其修改为使用Kafka斯库特（KafkaSpout）而不是普通的喷口。

当我运行应用程序时，kafka中的数据已经挂起，KafkaSpout的下一个线程不会收到任何消息——它进入，尝试在协调器下的空管理器列表上迭代，然后退出。

我的环境是一个相当旧的Cloudera虚拟机，有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

SpoutConfig spoutConfig = new SpoutConfig(new SpoutConfig.ZkHosts(zookeepers, "/brokers"),
        "gtest", // name of topic used by producer & consumer
        "/kafka",  // zookeeper root path for offset storing
        "KafkaSpout");

您正在使用"gtest"主题接收数据。请确保您是通过生产者发送来自此主题的数据。

在螺栓中，像那样打印元组

public void execute(Tuple tuple, BasicOutputCollector collector) {
        System.out.println(tuple);
    }

它应该用Kafka打印待处理的数据。

Answer 2

SpoutConfig spoutConf = new SpoutConfig(...)
spoutConf.startOffsetTime = kafka.api.OffsetRequest.LatestTime();

Answer 3

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

表格暴风-配置留档：

喷口中另一个非常有用的配置是强制喷口倒带到上一个偏移的能力。在喷口配置上执行forceStartOffsetTime，如下所示：

   spoutConfig.forceStartOffsetTime(-2);

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

你看起来怎么样？有一个片段会很有用。你可以用-2替换-1，看看你是否收到了什么，如果你的制作人很好，那么你应该可以消费。