Cleanup of error handling in AbstractOutputWriter.

.gitignore

@@ -15,3 +15,5 @@ hydra.ipr
 hydra.iws
 hydra-uber/src/main/resources/hydra-git.properties
 *.pyc
+hydra-main/web/spawn2/css/main.min.css
+hydra-main/web/spawn2/js/main.min.js

README.mdown

@@ -56,8 +56,15 @@ adding the [fl-cow](http://xmailserver.org/flcow.html) library to
 
 Many components assume that there is a local user called `hydra` and
 that all minion nodes can ssh as that user to each other.  This is
-used most prominently for `rsync` based replicas.
+used most prominently for `rsync` based replicas. The user `hydra`
+is not necessary when running a local-stack environment (see below).
 
+### OS X
+
+On OS X several utilities are necessary to run the local-stack environment:
+
+    brew install coreutils
+    brew install wget
 
 ## Components
 

hydra-avro/pom.xml

@@ -16,7 +16,7 @@
   <parent>
     <groupId>com.addthis.hydra</groupId>
     <artifactId>hydra-parent</artifactId>
-    <version>4.1.1-SNAPSHOT</version>
+    <version>4.1.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>hydra-avro</artifactId>

hydra-data/pom.xml

@@ -16,7 +16,7 @@
   <parent>
     <groupId>com.addthis.hydra</groupId>
     <artifactId>hydra-parent</artifactId>
-    <version>4.1.1-SNAPSHOT</version>
+    <version>4.1.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>hydra-data</artifactId>

hydra-data/src/main/java/com/addthis/hydra/data/query/AbstractQueryOp.java

@@ -59,9 +59,13 @@ public void close() throws IOException {
     }
 
     @Override
-    public void sendTable(DataTable table) {
+    public void sendTable(DataTable table, QueryStatusObserver queryStatusObserver) {
         for (Bundle row : table) {
-            send(row);
+            if (queryStatusObserver.queryCompleted || queryStatusObserver.queryCancelled) {
+                break;
+            } else {
+                send(row);
+            }
         }
         sendComplete();
     }

hydra-data/src/main/java/com/addthis/hydra/data/query/AbstractTableOp.java

@@ -28,9 +28,11 @@ public abstract class AbstractTableOp extends AbstractQueryOp implements QueryTa
     protected DataTable table;
 
     private final DataTableFactory tableFactory;
+    private final QueryStatusObserver queryStatusObserver;
 
-    public AbstractTableOp(DataTableFactory tableFactory) {
+    public AbstractTableOp(DataTableFactory tableFactory, QueryStatusObserver queryStatusObserver) {
         this.tableFactory = tableFactory;
+        this.queryStatusObserver = queryStatusObserver;
     }
 
     public abstract DataTable tableOp(DataTable table);
@@ -66,7 +68,7 @@ public void sendTable(DataTable table) {
     private void complete() {
         QueryOp next = getNext();
         if (next != null) {
-            next.sendTable(getTable());
+            next.sendTable(getTable(), queryStatusObserver);
         }
     }
 }

hydra-data/src/main/java/com/addthis/hydra/data/query/QueryOp.java

@@ -61,7 +61,7 @@ public interface QueryOp extends BundleOutput, Closeable {
     /**
      * @return true if this is an instance of table op
      */
-    public void sendTable(DataTable table);
+    public void sendTable(DataTable table, QueryStatusObserver queryStatusObserver);
 
     /**
      * @return simple debugging name

hydra-data/src/main/java/com/addthis/hydra/data/query/QueryOpProcessor.java

@@ -226,7 +226,7 @@ private void parseOps(String... opslist) {
                         appendOp(new OpRoll.AvgOpRoll(args));
                         break;
                     case CHANGEPOINTS:
-                        appendOp(new OpChangePoints(this, args));
+                        appendOp(new OpChangePoints(this, args, queryStatusObserver));
                         break;
                     case COMPARE:
                         appendOp(new OpCompare(args));
@@ -244,10 +244,10 @@ private void parseOps(String... opslist) {
                         appendOp(new OpDePivot(this, args));
                         break;
                     case DIFF:
-                        appendOp(new OpDiff(this, args));
+                        appendOp(new OpDiff(this, args, queryStatusObserver));
                         break;
                     case DISORDER:
-                        appendOp(new OpDisorder(this, args));
+                        appendOp(new OpDisorder(this, args, queryStatusObserver));
                         break;
                     case DSORT:
                         appendOp(new OpDiskSort(args, TMP_SORT_DIR_STRING, queryStatusObserver));
@@ -259,16 +259,16 @@ private void parseOps(String... opslist) {
                         appendOp(new OpFold(args));
                         break;
                     case FREQUENCYTABLE:
-                        appendOp(new OpFrequencyTable(this, args));
+                        appendOp(new OpFrequencyTable(this, args, queryStatusObserver));
                         break;
                     case GATHER:
                         appendOp(new OpGather(args, memTip, rowTip, tempDir.getPath(), queryStatusObserver));
                         break; // TODO move OpTop code into OpGather and delete OpTop
                     case HISTOGRAM:
-                        appendOp(new OpHistogram(args));
+                        appendOp(new OpHistogram(args, queryStatusObserver));
                         break;
                     case DISTRIBUTION:
-                        appendOp(new OpPercentileDistribution(this, args));
+                        appendOp(new OpPercentileDistribution(this, args, queryStatusObserver));
                         break;
                     case LIMIT:
                         appendOp(new OpLimit(args, queryStatusObserver));
@@ -283,7 +283,7 @@ private void parseOps(String... opslist) {
                         appendOp(new OpRoll.MaxOpRoll(args));
                         break;
                     case MEDIAN:
-                        appendOp(new OpMedian(this));
+                        appendOp(new OpMedian(this, queryStatusObserver));
                         break;
                     case MERGE:
                         appendOp(new OpMerge(args, queryStatusObserver));
@@ -307,25 +307,25 @@ private void parseOps(String... opslist) {
                         appendOp(new OpFill(args, true));
                         break;
                     case PERCENTRANK:
-                        appendOp(new OpPercentileRank(this, args));
+                        appendOp(new OpPercentileRank(this, args, queryStatusObserver));
                         break;
                     case PIVOT:
-                        appendOp(new OpPivot(this, args));
+                        appendOp(new OpPivot(this, args, queryStatusObserver));
                         break;
                     case RANGE:
-                        appendOp(new OpRange(this, args));
+                        appendOp(new OpRange(this, args, queryStatusObserver));
                         break;
                     case REVERSE:
-                        appendOp(new OpReverse(this));
+                        appendOp(new OpReverse(this, queryStatusObserver));
                         break;
                     case RMSING:
-                        appendOp(new OpRemoveSingletons(this, args));
+                        appendOp(new OpRemoveSingletons(this, args, queryStatusObserver));
                         break;
                     case RND_FAIL:
                         appendOp(new OpRandomFail(args));
                         break;
                     case SEEN:
-                        appendOp(new OpSeen(this, args));
+                        appendOp(new OpSeen(this, args, queryStatusObserver));
                         break;
                     case SKIP:
                         appendOp(new OpSkip(args));
@@ -350,7 +350,7 @@ private void parseOps(String... opslist) {
                         appendOp(new OpGather(args, memTip, rowTip, tempDir.getPath(), queryStatusObserver));
                         break;
                     case TRANSPOSE:
-                        appendOp(new OpTranspose(this));
+                        appendOp(new OpTranspose(this, queryStatusObserver));
                         break;
                 }
             }

hydra-data/src/main/java/com/addthis/hydra/data/query/ResultAccumulator.java

@@ -18,8 +18,8 @@
 
 public class ResultAccumulator extends AbstractTableOp {
 
-    public ResultAccumulator(QueryOpProcessor processor) {
-        super(processor);
+    public ResultAccumulator(QueryOpProcessor processor, QueryStatusObserver queryStatusObserver) {
+        super(processor, queryStatusObserver);
     }
 
     @Override

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpChangePoints.java

@@ -26,6 +26,7 @@
 import com.addthis.bundle.util.BundleColumnBinder;
 import com.addthis.bundle.value.ValueFactory;
 import com.addthis.hydra.data.query.AbstractTableOp;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 import com.addthis.hydra.data.util.ChangePoint;
 import com.addthis.hydra.data.util.FindChangePoints;
 
@@ -46,8 +47,8 @@ public class OpChangePoints extends AbstractTableOp {
     int inactiveThreshold;
     int windowSize;
 
-    public OpChangePoints(DataTableFactory factory, String args) {
-        super(factory);
+    public OpChangePoints(DataTableFactory factory, String args, QueryStatusObserver queryStatusObserver) {
+        super(factory, queryStatusObserver);
         try {
             String[] opt = args.split(":");
             timeColumn = opt.length >= 1 ? Integer.parseInt(opt[0]) : 0;
@@ -57,10 +58,10 @@ public OpChangePoints(DataTableFactory factory, String args) {
             minZScore = opt.length >= 5 ? Double.parseDouble(opt[4]) : 1.5;
             inactiveThreshold = opt.length >= 6 ? Integer.parseInt(opt[5]) : 1;
             windowSize = opt.length >= 7 ? Integer.parseInt(opt[6]) : 5;
-            log.warn("Initiated changepoints with parameters " +
+            log.info("Initiated changepoints with parameters " +
                      Strings.join(new Object[]{valColumn, minChange, minRatio, minZScore, inactiveThreshold}, ","));
         } catch (Exception ex)  {
-            log.warn("", ex);
+            log.error("", ex);
         }
     }
 

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpDiff.java

@@ -22,6 +22,7 @@
 import com.addthis.bundle.value.ValueString;
 import com.addthis.hydra.data.query.AbstractTableOp;
 import com.addthis.hydra.data.query.QueryOpProcessor;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 
 
 /**
@@ -55,8 +56,8 @@ private static enum DiffOp {
 
     private ColumnType type[];
 
-    public OpDiff(DataTableFactory tableFactory, String args) {
-        super(tableFactory);
+    public OpDiff(DataTableFactory tableFactory, String args, QueryStatusObserver queryStatusObserver) {
+        super(tableFactory, queryStatusObserver);
         type = new ColumnType[args.length()];
         for (int i = 0; i < args.length(); i++) {
             switch (args.charAt(i)) {
@@ -80,8 +81,8 @@ public OpDiff(DataTableFactory tableFactory, String args) {
         }
     }
 
-    public OpDiff(QueryOpProcessor processor, ColumnType type[]) {
-        super(processor);
+    public OpDiff(QueryOpProcessor processor, ColumnType type[], QueryStatusObserver queryStatusObserver) {
+        super(processor, queryStatusObserver);
         this.type = type;
     }
 

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpDisorder.java

@@ -25,6 +25,7 @@
 import com.addthis.bundle.util.ValueUtil;
 import com.addthis.bundle.value.ValueFactory;
 import com.addthis.hydra.data.query.AbstractTableOp;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 
 
 /**
@@ -58,8 +59,8 @@ public class OpDisorder extends AbstractTableOp {
     private int secondary;
     private int frequency;
 
-    public OpDisorder(DataTableFactory tableFactory, String args) {
-        super(tableFactory);
+    public OpDisorder(DataTableFactory tableFactory, String args, QueryStatusObserver queryStatusObserver) {
+        super(tableFactory, queryStatusObserver);
         String[] split = args.split(":");
         if (split.length < 2 || split.length > 3) {
             throw new IllegalArgumentException("expected disorder=p:s[:f], got " + args);

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpFrequencyTable.java

@@ -30,6 +30,7 @@
 import com.addthis.hydra.data.query.AbstractTableOp;
 import com.addthis.hydra.data.query.QueryOpProcessor;
 
+import com.addthis.hydra.data.query.QueryStatusObserver;
 import com.google.common.base.Objects;
 
 /**
@@ -45,8 +46,8 @@ public class OpFrequencyTable extends AbstractTableOp {
     private boolean appendTotal = false;
 
     // foo=0,1,2,3:4,5:0.99,p12
-    public OpFrequencyTable(QueryOpProcessor processor, String args) {
-        super(processor);
+    public OpFrequencyTable(QueryOpProcessor processor, String args, QueryStatusObserver queryStatusObserver) {
+        super(processor, queryStatusObserver);
         String tuple[] = Strings.splitArray(args, ":");
         cols = Strings.splitArray(tuple[0], ",");
 

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpHistogram.java

@@ -21,6 +21,7 @@
 import com.addthis.bundle.util.ValueUtil;
 import com.addthis.bundle.value.ValueFactory;
 import com.addthis.hydra.data.query.AbstractRowOp;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 import com.addthis.hydra.data.util.KeyHistogram;
 
 
@@ -39,6 +40,7 @@ public class OpHistogram extends AbstractRowOp {
 
     private final int scale;
     private final int column;
+    private final QueryStatusObserver queryStatusObserver;
 
     /**
      * usage: column, scale
@@ -48,7 +50,8 @@ public class OpHistogram extends AbstractRowOp {
      *
      * @param args
      */
-    public OpHistogram(String args) {
+    public OpHistogram(String args, QueryStatusObserver queryStatusObserver) {
+        this.queryStatusObserver = queryStatusObserver;
         int v[] = csvToInts(args);
         if (v.length < 1) {
             throw new RuntimeException("missing required column");
@@ -77,10 +80,14 @@ public Bundle rowOp(Bundle row) {
     public void sendComplete() {
         Map<Long, Long> map = histo.getSortedHistogram();
         for (Entry<Long, Long> e : map.entrySet()) {
-            Bundle row = rowFactory.createBundle();
-            binder.appendColumn(row, ValueFactory.create(e.getKey()));
-            binder.appendColumn(row, ValueFactory.create(e.getValue()));
-            getNext().send(row);
+            if (queryStatusObserver.queryCompleted || queryStatusObserver.queryCancelled) {
+                break;
+            } else {
+                Bundle row = rowFactory.createBundle();
+                binder.appendColumn(row, ValueFactory.create(e.getKey()));
+                binder.appendColumn(row, ValueFactory.create(e.getValue()));
+                getNext().send(row);
+            }
         }
         super.sendComplete();
     }

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpLimit.java

@@ -68,10 +68,10 @@ protected void setup(int limit, int offset) {
 
     @Override
     public void send(Bundle row) throws DataChannelError {
-        if (queryStatusObserver.queryCompleted ) {
+        if (queryStatusObserver.queryCompleted) {
             // Someone is attempting to send data even after we marked the query completed to true flag. This means
             // they are doing work and sending us bundles. Throw an exception because that needs to be checked.
-            log.error("Limit reached, sendComplete was called.");
+            log.trace("Limit reached, sendComplete was called.");
             return;
         }
 

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpMedian.java

@@ -16,6 +16,7 @@
 import com.addthis.bundle.table.DataTable;
 import com.addthis.hydra.data.query.AbstractTableOp;
 import com.addthis.hydra.data.query.QueryOpProcessor;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 
 
 /**
@@ -33,8 +34,8 @@
  */
 public class OpMedian extends AbstractTableOp {
 
-    public OpMedian(QueryOpProcessor processor) {
-        super(processor);
+    public OpMedian(QueryOpProcessor processor, QueryStatusObserver queryStatusObserver) {
+        super(processor, queryStatusObserver);
     }
 
     @Override

hydra-data/src/main/java/com/addthis/hydra/data/query/op/OpPercentileDistribution.java

@@ -20,6 +20,7 @@
 import com.addthis.bundle.util.ValueUtil;
 import com.addthis.bundle.value.ValueFactory;
 import com.addthis.hydra.data.query.AbstractTableOp;
+import com.addthis.hydra.data.query.QueryStatusObserver;
 import com.addthis.hydra.data.util.KeyPercentileDistribution;
 
 import com.yammer.metrics.stats.Snapshot;
@@ -49,8 +50,8 @@ public class OpPercentileDistribution extends AbstractTableOp {
      * @param tableFactory
      * @param args
      */
-    public OpPercentileDistribution(DataTableFactory tableFactory, String args) {
-        super(tableFactory);
+    public OpPercentileDistribution(DataTableFactory tableFactory, String args, QueryStatusObserver queryStatusObserver) {
+        super(tableFactory, queryStatusObserver);
         int v[] = csvToInts(args);
         if (v.length < 1) {
             throw new RuntimeException("missing required column");