Contre-pression dans la tempête

Question

Dans notre application tempête 1.0.2, nous sommes confrontés à des exceptions de mémoire insuffisante. Au débogage, nous avons vu que le bec de Kafka émettait trop de messages aux boulons. Les boulons fonctionnaient à une capacité de presque 4,0. Il y a donc un moyen de permettre la contre-pression dans la tempête de sorte que le bec s'écoule en fonction de la capacité dans les boulons. J'ai essayé d'activer la topology.backpressure.enable à true, mais j'ai couru vers ce numéro https://issues.apache.org/jira/browse/STORM-1949. Nous utilisons l'implémentation de KafkaSpout prête à l'emploi et étendons le BaseRichBolt pour nos boulons. Notre DAG est linéaire.

Answer 1

Vous pouvez gérer la pression arrière du KafkaSpout en réglant la maxSpoutPending valeur dans la configuration de la topologie,

Config config = new Config(); 
config.setMaxSpoutPending(200); 
config.setMessageTimeoutSecs(100); 

StormSubmitter.submitTopology("testtopology", config, builder.createTopology()); 

maxSpoutPending est le nombre de tuples qui peuvent être en attente accusé de réception dans votre topologie à un moment donné . En définissant cette propriété, le KafkaSpout sera intime pour ne plus consommer de données de Kafka à moins que le nombre de tuple non acquitté soit inférieur à la valeur maxSpoutPending.