阿帕奇风暴卡夫卡喷口滞后问题

Question

我正在使用 Storm 1.1.2 和 Kafka 0.11 构建一个 Java Spring 应用程序，以便在 Docker 容器中启动。

我的拓扑中的所有内容都按计划工作，但在 Kafka 的高负载下，Kafka 滞后随着时间的推移越来越多。

My KafkaSpoutConfig：

KafkaSpoutConfig<String,String> spoutConf = 
KafkaSpoutConfig.builder("kafkaContainerName:9092", "myTopic")
.setProp(ConsumerConfig.GROUP_ID_CONFIG, "myGroup")
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyObjectDeserializer.class)
.build()

那么我的拓扑如下

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("stormKafkaSpout", new KafkaSpout<String,String>(spoutConf), 25);
builder.setBolt("routerBolt", new RouterBolt(),25).shuffleGrouping("stormKafkaSpout");
Config conf = new Config();
conf.setNumWorkers(10);
conf.put(Config.STORM_ZOOKEEPER_SERVERS, ImmutableList.of("zookeeper"));
conf.put(Config.STORM_ZOOKEEPER_PORT, 2181);
conf.put(Config.NIMBUS_SEEDS, ImmutableList.of("nimbus"));
conf.put(Config.NIMBUS_THRIFT_PORT, 6627);
System.setProperty("storm.jar", "/opt/storm.jar");
StormSubmitter.submitTopology("topologyId", conf, builder.createTopology());

RouterBolt(扩展了BaseRichBolt(执行一个非常简单的switch语句，然后使用本地KafkaProducer对象向另一个主题发送新消息。就像我说的，一切都编译了，拓扑按预期运行，但在高负载(3000 条消息/秒(下，Kafka 滞后只是堆积起来，相当于拓扑的低吞吐量。

我试过禁用 acing

conf.setNumAckers(0);

和

conf.put(Config.TOPOLGY_ACKER_EXECUTORS, 0);

但我想这不是一个令人讨厌的问题。

我在 Storm UI 上看到 RouterBolt 在高负载下的执行延迟为 1.2ms，进程延迟为 .03ms，这让我相信 Spout 是瓶颈。并行提示也是 25，因为"myTopic"有 25 个分区。谢谢！

Answer 1

您可能会受到 https://issues.apache.org/jira/browse/STORM-3102 的影响，这会导致喷口在每次发射时执行非常昂贵的调用。请尝试升级到其中一个固定版本。

编辑：修复程序实际上尚未发布。您可能仍然想尝试通过使用例如 https://github.com/apache/storm/tree/1.1.x-branch 从源代码构建 spout 来构建 1.1.4 快照来尝试修复。