[
https://issues.apache.org/jira/browse/STORM-136?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rick Kellogg updated STORM-136:
-------------------------------
Component/s: storm-core
> [Trident] Bug when working with two spouts with stateful operator
> -----------------------------------------------------------------
>
> Key: STORM-136
> URL: https://issues.apache.org/jira/browse/STORM-136
> Project: Apache Storm
> Issue Type: Bug
> Components: storm-core
> Reporter: James Xu
> Priority: Minor
>
> https://github.com/nathanmarz/storm/issues/711
> Hi,
> I'm working with Trident and basically what i am trying to do is a windowed
> join across batches using partitionPersist and stateQuery on two different
> stream that come from TWO DIFFERENT SPOUTS.
> Both the spouts implement the IBatchSpout interface.
> The error i get is a NPE on StateQueryProcessor or on
> PartitionPersistProcessor depending on which one of the two spouts starts
> early.
> I try to debug this and what i have understand is that Trident use the same
> BatchID(txid) for the two different spouts and this take to a wrong
> initialization of state in the core processing nodes.
> If i use the same throughput for the two spout and i make one spout starts
> with a delay the problem doesn't occur (we don't have an overlap between the
> BachID(txid) of the different spouts).
> ----------
> liesrock: Sorry for the delay.
> I send to you the source code.
> https://dl.dropboxusercontent.com/u/49470846/TridentBug.tar.gz
> In this example i am using two spout:
> 1) the first one starts emitting tuples at a certain rate.
> 2) the second one is delayed by 5 seconds and emits at an higher rate than
> the first one.
> I print on the sterr some debug info, the most important is the Batch ID
> (txid).
> I print on the stdout info about the state window.
> You can see that this is exactly the case that i was explain in my first
> message.
> Thank you for your attention, i can't figure out on this, so i left Trident
> and i start working with Storm(i didn't have problem in implementing this on
> Storm).
> Let me know what is your opinion about it.
> Luca Muratore
> ----------
> xumingming: @liesrock try to minimize your test case so it can reproduce the
> issue but contains the least source files(preferably only one source file),
> the least lines of source code --- that makes us more easier to see whether
> the issue is in your app code or in storm core.
> ----------
> liesrock: As you requested, in the following link you can find the test case
> in one source file and with the minimum lines of code:
> https://dl.dropboxusercontent.com/u/49470846/SimpleTridentBug.zip
> Thanks for your patience.
> ----------
> xumingming: paste the case source code here to make it easier for others to
> follow:
> package storm.starter.trident;
> import java.io.IOException;
> import java.util.ArrayList;
> import java.util.List;
> import java.util.Map;
> import storm.trident.Stream;
> import storm.trident.TridentState;
> import storm.trident.TridentTopology;
> import storm.trident.operation.TridentCollector;
> import storm.trident.spout.IBatchSpout;
> import storm.trident.state.BaseQueryFunction;
> import storm.trident.state.BaseStateUpdater;
> import storm.trident.state.State;
> import storm.trident.state.StateFactory;
> import storm.trident.tuple.TridentTuple;
> import backtype.storm.Config;
> import backtype.storm.LocalCluster;
> import backtype.storm.task.IMetricsContext;
> import backtype.storm.task.TopologyContext;
> import backtype.storm.tuple.Fields;
> import backtype.storm.tuple.Values;
> public class BugTest {
> //Spout that simulates an ATM
> @SuppressWarnings("serial")
> public static class ATMSpout implements IBatchSpout {
> private int batchSize;
> private int initialSleepMilliTime;
> private int rate;
> private String name;
> private List<String> currentCCIDList;
> private long withdrawalID = 0;
> private final static String[] LOCATIONS = { "Madrid", "Barcelona",
> "Siviglia", "Granada", "Toledo", "Ibiza", "Valladolid", "Valencia" };
> public ATMSpout(int batchSize, int initialSleepMilliTime, int rate,
> String name) throws IOException {
> this.batchSize = batchSize;
> this.initialSleepMilliTime = initialSleepMilliTime;
> this.rate = rate;
> this.name = name;
> }
> //generate a list of "size" CCID
> private List<String> generateCCID(int size){
> //check the input parameter
> if(size < 0){
> System.err.println("Negative CCID list size");
> return null;
> }
> //initialize some variables
> List<String> aux = new ArrayList<String>();
> Integer randDigit = 0;
> String currentCCID = "";
> //create a random CCID
> for(int i = 0; i < size; i++){
> for(int j = 0; j < 16; j++){
> randDigit = (int)(Math.random() * 10);
> currentCCID += randDigit;
> }
> aux.add(currentCCID);
> currentCCID = "";
> }
> return aux;
> }
> @SuppressWarnings("rawtypes")
> @Override
> public void open(Map conf, TopologyContext context) {
> System.err.println("Open Spout instance");
> this.currentCCIDList = generateCCID(10000);
> //initial delay
> try {
> Thread.sleep(initialSleepMilliTime);
> } catch (InterruptedException e) {
> e.printStackTrace();
> }
> }
> @Override
> public void emitBatch(long batchId, TridentCollector collector) {
> System.err.println(name + " ---> Starting emitting Batch number :
> " + batchId);
> for(int i = 0; i < batchSize; i++) {
> try {
> Thread.sleep(rate);
> } catch (InterruptedException e) {
> e.printStackTrace();
> }
> collector.emit(getNextWithdrawal());
> }
> }
> @Override
> public void ack(long batchId) {
> }
> @Override
> public void close() {
> }
> @SuppressWarnings("rawtypes")
> @Override
> public Map getComponentConfiguration() {
> return new Config();
> }
> @Override
> public Fields getOutputFields() {
> return new Fields("withdrawalID", "ccID", "location", "amount",
> "timestamp");
> }
> private Values getNextWithdrawal(){
> int randIndexCCID = (int) ((Math.random()) *
> this.currentCCIDList.size());
> int randAmmount = (int) ((Math.random()) * 1000);
> int randIndexLocation = (int) ((Math.random()) *
> LOCATIONS.length);
> return new Values(++withdrawalID,
> this.currentCCIDList.get(randIndexCCID),
> LOCATIONS[randIndexLocation],
> randAmmount,
> System.currentTimeMillis());
> }
> }
> //state updater
> @SuppressWarnings("serial")
> private static class Updater extends BaseStateUpdater<DB> {
> private String streamName;
> public Updater(String streamName) {
> this.streamName = streamName;
> }
> @Override
> public void updateState(DB state, List<TridentTuple> tuples,
> TridentCollector collector) {
> System.err.println(streamName + " Update State");
> }
> }
> //query
> @SuppressWarnings("serial")
> public static class Query extends BaseQueryFunction<DB, TridentTuple> {
> private String streamName;
> public Query(String streamName) {
> this.streamName = streamName;
> }
> @Override
> public List<TridentTuple> batchRetrieve(DB state, List<TridentTuple>
> inputs) {
> System.err.println(streamName + " Query");
> List<TridentTuple> retList = new ArrayList<TridentTuple>();
> //return list
> for(int i = 0; i < 5; i++){
> retList.add(null);
> }
> return retList;
> }
> @Override
> public void execute(TridentTuple tuple, TridentTuple result,
> TridentCollector collector) {
> }
> }
> //state
> public static class DB implements State {
> @Override
> public void beginCommit(Long txid) {
> }
> @Override
> public void commit(Long txid) {
> }
> }
> //factory
> @SuppressWarnings("serial")
> public static class Factory implements StateFactory {
> @SuppressWarnings("rawtypes")
> @Override
> public State makeState(Map conf, IMetricsContext metrics, int
> partitionIndex, int numPartitions) {
> return new DB();
> }
> }
> public static void main(String[] args) throws Exception {
> Config conf = new Config();
> LocalCluster cluster = new LocalCluster();
> final int BATCH_SIZE = 5;
> final int DELAY_1 = 100;
> final int RATE_1 = 300;
> final int DELAY_2 = 5000;
> final int RATE_2 = 100;
> ATMSpout spout1 = new ATMSpout(BATCH_SIZE, DELAY_1, RATE_1,
> "SpoutLowRate");
> ATMSpout spout2 = new ATMSpout(BATCH_SIZE, DELAY_2, RATE_2,
> "SpoutHighRate");
> TridentTopology topology = new TridentTopology();
> Stream s1 = topology.newStream("stream1", spout1);
> Stream s2 = topology.newStream("stream2", spout2);
> TridentState leftState = s1.partitionPersist(new Factory(),
> s1.getOutputFields(),
> new Updater("left"));
> s2.stateQuery(leftState,
> s1.getOutputFields(),
> new Query("right"),
> new Fields("out"));
> cluster.submitTopology("BugTest", conf, topology.build());
> Thread.sleep(20000);
> cluster.shutdown();
> }
> }
> ----------
> xumingming: stacktrace:
> 5320 [Thread-27-b-0] ERROR backtype.storm.util - Async loop died!
> java.lang.RuntimeException: java.lang.NullPointerException
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:90)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:61)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$fn__3496$fn__3508$fn__3555.invoke(executor.clj:730)
> ~[storm-core-0.9.0-rc3.jar:na]
> at backtype.storm.util$async_loop$fn__442.invoke(util.clj:403)
> ~[storm-core-0.9.0-rc3.jar:na]
> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.4.0.jar:na]
> at java.lang.Thread.run(Thread.java:680) [na:1.6.0_29]
> Caused by: java.lang.NullPointerException: null
> at
> storm.trident.planner.processor.StateQueryProcessor.execute(StateQueryProcessor.java:69)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.planner.SubtopologyBolt$InitialReceiver.receive(SubtopologyBolt.java:194)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.planner.SubtopologyBolt.execute(SubtopologyBolt.java:130)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:355)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$fn__3496$tuple_action_fn__3498.invoke(executor.clj:615)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$mk_task_receiver$fn__3419.invoke(executor.clj:383)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.disruptor$clojure_handler$reify__2960.onEvent(disruptor.clj:43)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87)
> ~[storm-core-0.9.0-rc3.jar:na]
> ... 6 common frames omitted
> 5320 [Thread-27-b-0] ERROR backtype.storm.daemon.executor -
> java.lang.RuntimeException: java.lang.NullPointerException
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:90)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:61)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$fn__3496$fn__3508$fn__3555.invoke(executor.clj:730)
> ~[storm-core-0.9.0-rc3.jar:na]
> at backtype.storm.util$async_loop$fn__442.invoke(util.clj:403)
> ~[storm-core-0.9.0-rc3.jar:na]
> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.4.0.jar:na]
> at java.lang.Thread.run(Thread.java:680) [na:1.6.0_29]
> Caused by: java.lang.NullPointerException: null
> at
> storm.trident.planner.processor.StateQueryProcessor.execute(StateQueryProcessor.java:69)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.planner.SubtopologyBolt$InitialReceiver.receive(SubtopologyBolt.java:194)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.planner.SubtopologyBolt.execute(SubtopologyBolt.java:130)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:355)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$fn__3496$tuple_action_fn__3498.invoke(executor.clj:615)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.daemon.executor$mk_task_receiver$fn__3419.invoke(executor.clj:383)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.disruptor$clojure_handler$reify__2960.onEvent(disruptor.clj:43)
> ~[storm-core-0.9.0-rc3.jar:na]
> at
> backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:87)
> ~[storm-core-0.9.0-rc3.jar:na]
> ... 6 common frames omitted
> 5334 [Thread-27-b-0] INFO backtype.storm.util - Halting process: ("Worker
> died")
> The NullPointerException is thrown because state is null:
> https://github.com/nathanmarz/storm/blob/master/storm-core/src/jvm/storm/trident/planner/processor/StateQueryProcessor.java?source=cc#L57-L61
> @Override
> public void execute(ProcessorContext processorContext, String streamId,
> TridentTuple tuple) {
> BatchState state = (BatchState)
> processorContext.state[_context.getStateIndex()];
> state.tuples.add(tuple);
> state.args.add(_projection.create(tuple));
> }
> Why state is null? state is initialized in startBatch method:
> https://github.com/nathanmarz/storm/blob/master/storm-core/src/jvm/storm/trident/planner/processor/StateQueryProcessor.java?source=cc#L52-L54
> public void startBatch(ProcessorContext processorContext) {
> processorContext.state[_context.getStateIndex()] = new BatchState();
> }
> In this case, startBatch is not called before execute is called.
> Why startBatch is not called? startBatch is called by initBatchState,
> initBatchState should be called by TridentBoltExecutor.execute:
> if(tracked==null) {
> tracked = new TrackedBatch(new BatchInfo(batchGroup, id,
> _bolt.initBatchState(batchGroup, id)), _coordConditions.get(batchGroup),
> id.getAttemptId());
> _batches.put(id.getId(), tracked);
> }
> In the code above, initBatchState is called with a batchGroup arg, when there
> are multiple batchGroups in one batch, the TrackedBatch will only be created
> once, then the initBatchState will only be called for the first batchGroup,
> then startBatch method of processors in the rest batchGroups will not be
> called, then the NullPointerException, @nathanmarz, is this a bug or just not
> using Trident in the correct way?
> ----------
> nathanmarz: Looks like a bug.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
