fix bug in LinearDRPCTopologyBuilder

Nathan Marz · Nathan Marz · commit 436a8acf4507 · 2011-11-30T19:04:17.000-08:00
diff --git a/src/jvm/backtype/storm/drpc/LinearDRPCTopologyBuilder.java b/src/jvm/backtype/storm/drpc/LinearDRPCTopologyBuilder.java
@@ -91,8 +91,14 @@ private StormTopology createTopology(DRPCSpout spout) {
             if(i==0 && component.declarations.size()==0) {
                 declarer.noneGrouping(PREPARE_ID, PrepareRequest.ARGS_STREAM);
             } else {
+                String prevId;
+                if(i==0) {
+                    prevId = PREPARE_ID;
+                } else {
+                    prevId = boltId(i-1);
+                }
                 for(InputDeclaration declaration: component.declarations) {
-                    declaration.declare(boltId(i-1), declarer);
+                    declaration.declare(prevId, declarer);
                 }
             }
             if(i>0) {
diff --git a/src/jvm/backtype/storm/drpc/PrepareRequest.java b/src/jvm/backtype/storm/drpc/PrepareRequest.java
@@ -9,10 +9,11 @@
 import backtype.storm.tuple.Values;
 import java.util.Map;
 import java.util.Random;
+import backtype.storm.utils.Utils;
 
 
 public class PrepareRequest implements IBasicBolt {
-    public static final String ARGS_STREAM = "args";
+    public static final String ARGS_STREAM = Utils.DEFAULT_STREAM_ID;
     public static final String RETURN_STREAM = "ret";
     public static final String ID_STREAM = "id";
 
