[jira] [Updated] (STORM-124) Tuple Lost in Trident Topology on Remote Cluster, while the same topology works correctly on local cluster

["Rick Kellogg (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/STORM-124?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Rick Kellogg updated STORM-124:
-------------------------------
    Component/s: storm-core

> Tuple Lost in Trident Topology on Remote Cluster, while the same topology works correctly on local cluster
> ----------------------------------------------------------------------------------------------------------
>
>                 Key: STORM-124
>                 URL: https://issues.apache.org/jira/browse/STORM-124
>             Project: Apache Storm
>          Issue Type: Bug
>          Components: storm-core
>            Reporter: James Xu
>            Priority: Minor
>
> https://github.com/nathanmarz/storm/issues/535
> I have created a trident topology, while it runs correctly on local cluster , it gets random result when deployed on remote cluster due to tuple loss in trident
> Code example is as followed, In ExtractFunction it will parse kafka message and create 4 tuples from the content(5 fields), then CustomReduceAggregator will aggregate based on first 4 fields.
> I generated 800 kafka messages during the test, ExtractFunction is expected to get 800 tuples and generate 800*4=3200 tuples for CustomReduceAggregator.
> This works correctly in local cluster.
> In remoted cluster, while ExtractFunction generated 3200 tuples, CustomReduceAggregator will always get only a few hundred tuples, therefore the result is not correct
> Is this a bug in storm/trident or there is some configuration needed for my trident topology?
>     TridentKafkaConfig tridentKafkaConfig = new TridentKafkaConfig(new ZkHosts(zkServer+":"+zkPort, zkBrokerPath), kafkaTopic);
>     tridentKafkaConfig.forceStartOffsetTime(-2);
>     TridentTopology topology=new TridentTopology();
>     TridentState tridentState = topology.newStream("kafkaSpout"+System.currentTimeMillis(), new TransactionalTridentKafkaSpout(tridentKafkaConfig))
>         .parallelismHint(1)
>         .each(new Fields("bytes"), new ExtractFunction(), new Fields("a", "b", "c", "b", "e"))
>         .parallelismHint(1)
>         .groupBy(new Fields("a", "b", "c", "f"))
>         .persistentAggregate(new CassandraMapState.Factory(StateType.TRANSACTIONAL, "cf_1"),  
>                 new Fields("a", "b", "c", "b", "e"), 
>                 new CustomReduceAggregator(), 
>                 new Fields("value"))
>         .parallelismHint(1);
>     topology.newDRPCStream("profile", localDRPC)
>         .each(new Fields("args"), new DrpcParamSplitFunction(), new Fields("a", "b", "c", "b"))
>         .groupBy(new Fields("a", "b", "c", "b"))
>         .stateQuery(tridentState, new Fields("a", "b", "c", "b"), new MapGet(), new Fields("value"));
>     return topology.build();



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)