HBASE-1780 HTable.flushCommits clears write buffer in finally clause

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hbase/trunk@808579 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -357,6 +357,7 @@ Release 0.20.0 - Unreleased
                new updates into the 'kvset' (active part)
    HBASE-1767  test zookeeper broken in trunk and 0.20 branch; broken on
                hudson too
+   HBASE-1780  HTable.flushCommits clears write buffer in finally clause
 
   IMPROVEMENTS
    HBASE-1089  Add count of regions on filesystem to master UI; add percentage

src/java/org/apache/hadoop/hbase/client/HConnection.java

@@ -189,6 +189,6 @@ public interface HConnection {
    * @param tableName The name of the table
    * @throws IOException
    */
-  public void processBatchOfRows(ArrayList<Put> list, byte[] tableName)
+  public int processBatchOfRows(ArrayList<Put> list, byte[] tableName)
       throws IOException;
 }

src/java/org/apache/hadoop/hbase/client/HConnectionManager.java

@@ -997,10 +997,10 @@ public class HConnectionManager implements HConstants {
       return location;
     }
 
-    public void processBatchOfRows(ArrayList<Put> list, byte[] tableName)
+    public int processBatchOfRows(ArrayList<Put> list, byte[] tableName)
         throws IOException {
       if (list.isEmpty()) {
-        return;
+        return 0;
       }
       boolean retryOnlyOne = false;
       if (list.size() > 1) {
@@ -1014,7 +1014,8 @@ public class HConnectionManager implements HConstants {
       byte [] region = currentRegion;
       boolean isLastRow = false;
       Put [] putarray = new Put[0];
-      for (int i = 0, tries = 0; i < list.size() && tries < this.numRetries; i++) {
+      int i, tries;
+      for (i = 0, tries = 0; i < list.size() && tries < this.numRetries; i++) {
         Put put = list.get(i);
         currentPuts.add(put);
         // If the next Put goes to a new region, then we are to clear
@@ -1072,6 +1073,7 @@ public class HConnectionManager implements HConstants {
           currentPuts.clear();
         }
       }
+      return i;
     }
 
     void close(boolean stopProxy) {

src/java/org/apache/hadoop/hbase/client/HTable.java

@@ -54,14 +54,16 @@ import org.apache.hadoop.hbase.util.Writables;
 
 
 /**
- * Used to communicate with a single HBase table
+ * Used to communicate with a single HBase table.
+ * <p>
+ * This class is not MT safe for writes.
  */
 public class HTable implements HTableInterface {
   private final HConnection connection;
   private final byte [] tableName;
   protected final int scannerTimeout;
   private volatile HBaseConfiguration configuration;
-  private ArrayList<Put> writeBuffer;
+  private final ArrayList<Put> writeBuffer = new ArrayList<Put>();
   private long writeBufferSize;
   private boolean autoFlush;
   private long currentWriteBufferSize;
@@ -121,7 +123,6 @@ public class HTable implements HTableInterface {
       conf.getInt("hbase.regionserver.lease.period", 60 * 1000);
     this.configuration = conf;
     this.connection.locateRegion(tableName, HConstants.EMPTY_START_ROW);
-    this.writeBuffer = new ArrayList<Put>();
     this.writeBufferSize = conf.getLong("hbase.client.write.buffer", 2097152);
     this.autoFlush = true;
     this.currentWriteBufferSize = 0;
@@ -578,11 +579,15 @@ public class HTable implements HTableInterface {
    * @throws IOException
    */
   public void flushCommits() throws IOException {
+    int last = 0;
     try {
-      connection.processBatchOfRows(writeBuffer, tableName);
+      last = connection.processBatchOfRows(writeBuffer, tableName);
     } finally {
+      writeBuffer.subList(0, last).clear();
       currentWriteBufferSize = 0;
-      writeBuffer.clear();
+      for (int i = 0; i < writeBuffer.size(); i++) {
+        currentWriteBufferSize += writeBuffer.get(i).heapSize();
+      }
     }
   }