bshakur8
diff --git a/‎tools/docker/Dockerfile renamed to ‎Dockerfile
Lines changed: 4 additions & 4 deletions b/‎tools/docker/Dockerfile renamed to ‎Dockerfile
Lines changed: 4 additions & 4 deletions
diff --git a/‎Makefile.am
Lines changed: 3 additions & 0 deletions b/‎Makefile.am
Lines changed: 3 additions & 0 deletions
diff --git a/‎configure.ac
Lines changed: 2 additions & 1 deletion b/‎configure.ac
Lines changed: 2 additions & 1 deletion
diff --git a/‎src/core/IncomingDataPoints.java
Lines changed: 14 additions & 57 deletions b/‎src/core/IncomingDataPoints.java
Lines changed: 14 additions & 57 deletions
diff --git a/‎src/core/Query.java
Lines changed: 1 addition & 1 deletion b/‎src/core/Query.java
Lines changed: 1 addition & 1 deletion
diff --git a/‎src/core/RpcResponder.java
Lines changed: 110 additions & 0 deletions b/‎src/core/RpcResponder.java
Lines changed: 110 additions & 0 deletions
diff --git a/‎src/core/SaltScanner.java
Lines changed: 20 additions & 2 deletions b/‎src/core/SaltScanner.java
Lines changed: 20 additions & 2 deletions
@@ -2,7 +2,7 @@ FROM java:openjdk-8-alpine
 
 MAINTAINER [email protected]
 
-ENV       VERSION 2.3.0-RC1
+ENV       VERSION 2.5.0-SNAPSHOT
 ENV       WORKDIR /usr/share/opentsdb
 ENV       LOGDIR  /var/log/opentsdb
 ENV       DATADIR /data/opentsdb
@@ -32,10 +32,10 @@ ENV       TSDB_PORT  4244
 
 WORKDIR   $WORKDIR
 
-ADD       libs $WORKDIR/libs
-ADD       logback.xml $WORKDIR
+ADD       third_party/*/*.jar $WORKDIR/libs/
+ADD       src/logback.xml $WORKDIR
 ADD       tsdb-$VERSION.jar $WORKDIR
-ADD       opentsdb.conf $ETCDIR/opentsdb.conf
+ADD       src/opentsdb.conf $ETCDIR/opentsdb.conf
 
 VOLUME    ["/etc/openstsdb"]
 VOLUME    ["/data/opentsdb"]
 
@@ -82,6 +82,7 @@ tsdb_SRC := \
 	src/core/RequestBuilder.java	\
 	src/core/RowKey.java	\
 	src/core/RowSeq.java	\
+	src/core/RpcResponder.java	\
 	src/core/iRowSeq.java	\
 	src/core/SaltScanner.java	\
 	src/core/SeekableView.java	\
@@ -127,6 +128,7 @@ tsdb_SRC := \
 	src/query/expression/ExpressionReader.java	\
 	src/query/expression/Expressions.java	\
 	src/query/expression/ExpressionTree.java	\
+	src/query/expression/FirstDifference.java	\
 	src/query/expression/HighestCurrent.java	\
 	src/query/expression/HighestMax.java	\
 	src/query/expression/IntersectionIterator.java	\
@@ -323,6 +325,7 @@ test_SRC := \
 	test/core/TestRateSpan.java	\
 	test/core/TestRowKey.java	\
 	test/core/TestRowSeq.java	\
+	test/core/TestRpcResponsder.java	\
 	test/core/TestSaltScanner.java	\
 	test/core/TestSeekableViewChain.java	\
 	test/core/TestSpan.java	\
 
@@ -14,7 +14,8 @@
 # along with this library.  If not, see <http://www.gnu.org/licenses/>.
 
 # Semantic Versioning (see http://semver.org/).
-AC_INIT([opentsdb], [2.4.1], [[email protected]])
+AC_INIT([opentsdb], [2.5.0-SNAPSHOT], [[email protected]])
+
 AC_CONFIG_AUX_DIR([build-aux])
 AM_INIT_AUTOMAKE([foreign])
 
 
@@ -18,6 +18,7 @@
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 
 import com.stumbleupon.async.Callback;
 import com.stumbleupon.async.Deferred;
@@ -45,6 +46,11 @@ final class IncomingDataPoints implements WritableDataPoints {
    */
   static final Histogram putlatency = new Histogram(16000, (short) 2, 100);
 
+  /**
+   * Keep track of the number of UIDs that came back null with auto_metric disabled.
+   */
+  static final AtomicLong auto_metric_rejection_count = new AtomicLong();
+
   /** The {@code TSDB} instance we belong to. */
   private final TSDB tsdb;
 
@@ -137,9 +143,14 @@ static byte[] rowKeyTemplate(final TSDB tsdb, final String metric,
 
     short pos = (short) Const.SALT_WIDTH();
 
-    copyInRowKey(row, pos,
-        (tsdb.config.auto_metric() ? tsdb.metrics.getOrCreateId(metric)
-            : tsdb.metrics.getId(metric)));
+    byte[] metric_id  = (tsdb.config.auto_metric() ? tsdb.metrics.getOrCreateId(metric)
+            : tsdb.metrics.getId(metric));
+
+    if(!tsdb.config.auto_metric() && metric_id == null) {
+      auto_metric_rejection_count.incrementAndGet();
+    }
+
+    copyInRowKey(row, pos, metric_id);
     pos += metric_width;
 
     pos += Const.TIMESTAMP_BYTES;
@@ -151,60 +162,6 @@ static byte[] rowKeyTemplate(final TSDB tsdb, final String metric,
     return row;
   }
 
-  /**
-   * Returns a partially initialized row key for this metric and these tags. The
-   * only thing left to fill in is the base timestamp.
-   * 
-   * @since 2.0
-   */
-  static Deferred<byte[]> rowKeyTemplateAsync(final TSDB tsdb,
-      final String metric, final Map<String, String> tags) {
-    final short metric_width = tsdb.metrics.width();
-    final short tag_name_width = tsdb.tag_names.width();
-    final short tag_value_width = tsdb.tag_values.width();
-    final short num_tags = (short) tags.size();
-
-    int row_size = (Const.SALT_WIDTH() + metric_width + Const.TIMESTAMP_BYTES 
-        + tag_name_width * num_tags + tag_value_width * num_tags);
-    final byte[] row = new byte[row_size];
-
-    // Lookup or create the metric ID.
-    final Deferred<byte[]> metric_id;
-    if (tsdb.config.auto_metric()) {
-      metric_id = tsdb.metrics.getOrCreateIdAsync(metric, metric, tags);
-    } else {
-      metric_id = tsdb.metrics.getIdAsync(metric);
-    }
-
-    // Copy the metric ID at the beginning of the row key.
-    class CopyMetricInRowKeyCB implements Callback<byte[], byte[]> {
-      public byte[] call(final byte[] metricid) {
-        copyInRowKey(row, (short) Const.SALT_WIDTH(), metricid);
-        return row;
-      }
-    }
-
-    // Copy the tag IDs in the row key.
-    class CopyTagsInRowKeyCB implements
-        Callback<Deferred<byte[]>, ArrayList<byte[]>> {
-      public Deferred<byte[]> call(final ArrayList<byte[]> tags) {
-        short pos = (short) (Const.SALT_WIDTH() + metric_width);
-        pos += Const.TIMESTAMP_BYTES;
-        for (final byte[] tag : tags) {
-          copyInRowKey(row, pos, tag);
-          pos += tag.length;
-        }
-        // Once we've resolved all the tags, schedule the copy of the metric
-        // ID and return the row key we produced.
-        return metric_id.addCallback(new CopyMetricInRowKeyCB());
-      }
-    }
-
-    // Kick off the resolution of all tags.
-    return Tags.resolveOrCreateAllAsync(tsdb, metric, tags)
-        .addCallbackDeferring(new CopyTagsInRowKeyCB());
-  }
-
   public void setSeries(final String metric, final Map<String, String> tags) {
     checkMetricAndTags(metric, tags);
     try {
 
@@ -197,7 +197,7 @@ public Deferred<Object> configureFromQuery(final TSQuery query,
    * way we get this one data point is by aggregating all the data points of
    * that interval together using an {@link Aggregator}.  This enables you
    * to compute things like the 5-minute average or 10 minute 99th percentile.
-   * @param interval Number of seconds wanted between each data point.
+   * @param interval Number of milliseconds wanted between each data point.
    * @param downsampler Aggregation function to use to group data points
    * within an interval.
    */
 
@@ -0,0 +1,110 @@
+// This file is part of OpenTSDB.
+// Copyright (C) 2010-2017  The OpenTSDB Authors.
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 2.1 of the License, or (at your
+// option) any later version.  This program is distributed in the hope that it
+// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
+// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
+// General Public License for more details.  You should have received a copy
+// of the GNU Lesser General Public License along with this program.  If not,
+// see <http://www.gnu.org/licenses/>.
+package net.opentsdb.core;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import net.opentsdb.utils.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * This class is responsible for building result of requests and
+ * respond to clients asynchronously.
+ *
+ * It can reduce requests that stacking in AsyncHBase, especially put requests.
+ * When a HBase's RPC has completed, the "AsyncHBase I/O worker" just decodes
+ * the response, and then do callback by this class asynchronously. We should
+ * take up workers as short as possible time so that workers can remove RPCs
+ * from in-flight state more quickly.
+ *
+ */
+public class RpcResponder {
+
+  private static final Logger LOG = LoggerFactory.getLogger(RpcResponder.class);
+
+  public static final String TSD_RESPONSE_ASYNC_KEY = "tsd.core.response.async";
+  public static final boolean TSD_RESPONSE_ASYNC_DEFAULT = true;
+
+  public static final String TSD_RESPONSE_WORKER_NUM_KEY =
+          "tsd.core.response.worker.num";
+  public static final int TSD_RESPONSE_WORKER_NUM_DEFAULT = 10;
+
+  private final boolean async;
+  private ExecutorService responders;
+  private volatile boolean running = true;
+
+  RpcResponder(final Config config) {
+    async = config.getBoolean(TSD_RESPONSE_ASYNC_KEY,
+            TSD_RESPONSE_ASYNC_DEFAULT);
+
+    if (async) {
+      int threads = config.getInt(TSD_RESPONSE_WORKER_NUM_KEY,
+              TSD_RESPONSE_WORKER_NUM_DEFAULT);
+      responders = Executors.newFixedThreadPool(threads,
+        new ThreadFactoryBuilder()
+          .setNameFormat("OpenTSDB Responder #%d")
+          .setDaemon(true)
+          .setUncaughtExceptionHandler(new ExceptionHandler())
+          .build());
+    }
+
+    LOG.info("RpcResponder mode: {}", async ? "async" : "sync");
+  }
+
+  public void response(Runnable run) {
+    if (async) {
+      if (running) {
+        responders.execute(run);
+      } else {
+        throw new IllegalStateException("RpcResponder is closing or closed.");
+      }
+    } else {
+      run.run();
+    }
+  }
+
+  public void close() {
+    if (running) {
+      running = false;
+      responders.shutdown();
+    }
+
+    boolean completed;
+    try {
+      completed = responders.awaitTermination(5, TimeUnit.MINUTES);
+    } catch (InterruptedException e) {
+      completed = false;
+    }
+
+    if (!completed) {
+      LOG.warn(
+          "There are still some results that are not returned to the clients.");
+    }
+  }
+
+  public boolean isAsync() {
+    return async;
+  }
+
+  private class ExceptionHandler implements Thread.UncaughtExceptionHandler {
+    @Override
+    public void uncaughtException(Thread t, Throwable e) {
+      LOG.error("Run into an uncaught exception in thread: " + t.getName(), e);
+    }
+  }
+}
@@ -491,7 +491,8 @@ final class ScannerCB implements Callback<Object,
     private long rows_pre_filter = 0;
     private long dps_post_filter = 0;
     private long rows_post_filter = 0;
-    
+    private long query_timeout = tsdb.getConfig().getLong("tsd.query.timeout");
+
     public ScannerCB(final Scanner scanner, final int index) {
       this.scanner = scanner;
       this.index = index;
@@ -554,7 +555,24 @@ public Object call(final ArrayList<ArrayList<KeyValue>> rows)
         final List<Deferred<Object>> lookups = 
             filters != null && !filters.isEmpty() ? 
                 new ArrayList<Deferred<Object>>(rows.size()) : null;
-        
+
+        // fail the query when the timeout exceeded
+        if (this.query_timeout > 0 && fetch_time > (this.query_timeout * 1000000)) {
+          try {
+            close(false);
+            handleException(
+                new QueryException(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE,
+                "Sorry, your query timed out. Time limit: "
+                    + this.query_timeout + " ms, fetch time: "
+                    + (double)(fetch_time)/1000000 + " ms. Please try filtering "
+                    + "using more tags or decrease your time range."));
+            return false;
+          } catch (Exception e) {
+            LOG.error("Sorry, Scanner is closed: " + scanner, e);
+            return false;
+          }
+        }
+
         // validation checking before processing the next set of results. It's 
         // kinda funky but we want to allow queries to sneak through that were
         // just a *tad* over the limits so that's why we don't check at the