KafkaSpout прекратил излучать, через некоторое время появились неудачные кортежи
У меня есть проблема с Storm KafkaSpout. У меня есть тема "twitter_streaming" в Кафке с 3 разделами. Это моя топология:
SpoutConfig streamingTweetWriterConfig = new SpoutConfig(zkHosts, TwitterConstant.Kafka.TWITTER_STREAMING_TOPIC,
"/" + TwitterConstant.Kafka.TWITTER_STREAMING_TOPIC, TwitterConstant.StormComponentID.STREAMING_TWEET_WRITER_SPOUT);
streamingTweetWriterConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
streamingTweetWriterConfig.startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
KafkaSpout streamingTweetSpout = new KafkaSpout(streamingTweetWriterConfig);
builder.setSpout(TwitterConstant.StormComponentID.STREAMING_TWEET_WRITER_SPOUT, streamingTweetSpout, 3);
builder.setBolt(TwitterConstant.StormComponentID.STREAMING_TWEET_WRITER_BOLT, new StreamingTweetWriterBolt(), 3).shuffleGrouping(TwitterConstant.StormComponentID.STREAMING_TWEET_WRITER_SPOUT);
Map<String, Object> storm_conf = Utils.readStormConfig();
storm_conf.put(Config.NIMBUS_SEEDS, Arrays.asList(TwitterConstant.StormConfiguration.NIMBUS_HOST));
storm_conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 2048);
storm_conf.put(Config.TOPOLOGY_BACKPRESSURE_ENABLE, false);
storm_conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 16384);
storm_conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, 16384);
storm_conf.put(Config.TOPOLOGY_WORKERS, 3);
storm_conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 120);
Моя топология очень проста: просто KafkaSpout и Bolt для записи потоковых твитов в Kafka в базу данных (пока не нужно заботиться о сканировании твитов). Подсказки параллельности KafkaSpout и TweetWriterBolt равны 3. Но после примерно 30 тыс. Кортежей KafkaSpout прекратил отправлять данные в Bolt и начал давать сбой. Я пытался пару раз с различными конфигами из вопросов Stackru, но проблема все еще существует, KafkaSpout всегда останавливался, чтобы генерировать записи с 30K-битами.
Кто-нибудь знает, как решить эту проблему, или некоторые советуют отлаживать? Заранее спасибо. PS: я прикрепил скриншот Storm UI, Spout сказал, что переносит кортежи, но Bolt всегда получает только 30000 кортежей
