Я пытаюсь интегрировать MongoDB и Storm-Kafka, Kafka Producer производит данные из MongoDB, но не может получить все записи со стороны потребителя.Он потребляет только 500-600 записей из 1 миллиона записей.
В файле журнала нет ошибок, топология все еще жива, но дальнейшие записи не обрабатываются.
Версия Kafka: 0.10. * Stormверсия: 1.2.1
Нужно ли мне добавлять какие-либо конфиги в Consumer?
conf.put(Config.TOPOLOGY_BACKPRESSURE_ENABLE, false);
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 2048);
conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 16384);
conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, 16384);
BrokerHosts hosts = new ZkHosts(zookeeperUrl);
SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
spoutConfig.scheme = new KeyValueSchemeAsMultiScheme(new StringKeyValueScheme());
spoutConfig.fetchSizeBytes = 25000000;
if (startFromBeginning) {
spoutConfig.startOffsetTime = OffsetRequest.EarliestTime();
} else {
spoutConfig.startOffsetTime = OffsetRequest.LatestTime();
}
return new KafkaSpout(spoutConfig);
}
Я хочу, чтобы носик Кафки прочитал все записи из темы кафки, которые были созданы производителем.