Need better throttling logic during batch imports

[tsuna]

Need better throttling logic during batch imports

[tsuna]

Here's a quick attempt to mitigate the issue. Not sure yet whether that's good enough. It seems like it was enough for a 28M data-point batch import that was running at about 150k edits/s on average, but I'm afraid there was a bit of luck involved in that attempt (splits didn't take regions offline for very long).

diff --git a/src/tools/TextImporter.java b/src/tools/TextImporter.java
index 8fbba3b..5766bef 100644
--- a/src/tools/TextImporter.java
+++ b/src/tools/TextImporter.java
@@ -92,6 +92,7 @@ final class TextImporter {
   }

   static volatile boolean throttle = false;
+  static final Object throttle_condition = new Object();

   private static int importFile(final HBaseClient client,
                                 final TSDB tsdb,
@@ -109,8 +110,14 @@ final class TextImporter {
             LOG.warn("Need to throttle, HBase isn't keeping up.", e);
             throttle = true;
             final HBaseRpc rpc = e.getFailedRpc();
-            if (rpc instanceof PutRequest) {
-              client.put((PutRequest) rpc);  // Don't lose edits.
+            if (rpc instanceof PutRequest) {  // Don't lose edits.
+              client.put((PutRequest) rpc).addCallbacks(new Callback<Object, Object>() {
+                public Object call(final Object arg) {
+                  throttle = false;
+                  throttle_condition.notifyAll();
+                  return arg;
+                }
+              }, new Errback());
             }
             return null;
           }
@@ -165,9 +172,11 @@ final class TextImporter {
           LOG.info("Throttling...");
           long throttle_time = System.nanoTime();
           try {
-            d.joinUninterruptibly();
-          } catch (Exception e) {
-            throw new RuntimeException("Should never happen", e);
+            while (throttle) {
+              throttle_condition.wait();
+            }
+          } catch (InterruptedException e) {
+            throw new RuntimeException("interrupted", e);
           }
           throttle_time = System.nanoTime() - throttle_time;
           if (throttle_time < 1000000000L) {
@@ -175,7 +184,6 @@ final class TextImporter {
             try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException("interrupted", e); }
           }
           LOG.info("Done throttling...");
-          throttle = false;
         }
       }
     } catch (RuntimeException e) {