fix batchresult handle leak

Summary:
This is related to PR https://github.com/facebook/rocksdb/pull/1642

Sorry about this extra PR, as my workflow was messed up when I checked in another work item by accident.
Closes https://github.com/facebook/rocksdb/pull/1681

Differential Revision: D4379513

Pulled By: yiwu-arbug

fbshipit-source-id: a668d4c

java/src/main/java/org/rocksdb/AbstractWriteBatch.java

@@ -14,71 +14,60 @@ public abstract class AbstractWriteBatch extends RocksObject
 
   @Override
   public int count() {
-    assert (isOwningHandle());
     return count0(nativeHandle_);
   }
 
   @Override
   public void put(byte[] key, byte[] value) {
-    assert (isOwningHandle());
     put(nativeHandle_, key, key.length, value, value.length);
   }
 
   @Override
   public void put(ColumnFamilyHandle columnFamilyHandle, byte[] key,
       byte[] value) {
-    assert (isOwningHandle());
     put(nativeHandle_, key, key.length, value, value.length,
         columnFamilyHandle.nativeHandle_);
   }
 
   @Override
   public void merge(byte[] key, byte[] value) {
-    assert (isOwningHandle());
     merge(nativeHandle_, key, key.length, value, value.length);
   }
 
   @Override
   public void merge(ColumnFamilyHandle columnFamilyHandle, byte[] key,
       byte[] value) {
-    assert (isOwningHandle());
     merge(nativeHandle_, key, key.length, value, value.length,
         columnFamilyHandle.nativeHandle_);
   }
 
   @Override
   public void remove(byte[] key) {
-    assert (isOwningHandle());
     remove(nativeHandle_, key, key.length);
   }
 
   @Override
   public void remove(ColumnFamilyHandle columnFamilyHandle, byte[] key) {
-    assert (isOwningHandle());
     remove(nativeHandle_, key, key.length, columnFamilyHandle.nativeHandle_);
   }
 
   @Override
   public void putLogData(byte[] blob) {
-    assert (isOwningHandle());
     putLogData(nativeHandle_, blob, blob.length);
   }
 
   @Override
   public void clear() {
-    assert (isOwningHandle());
     clear0(nativeHandle_);
   }
 
   @Override
   public void setSavePoint() {
-    assert (isOwningHandle());
     setSavePoint0(nativeHandle_);
   }
 
   @Override
   public void rollbackToSavePoint() throws RocksDBException {
-    assert (isOwningHandle());
     rollbackToSavePoint0(nativeHandle_);
   }
 

java/src/main/java/org/rocksdb/TransactionLogIterator.java

@@ -76,7 +76,7 @@ public class TransactionLogIterator extends RocksObject {
     public BatchResult(final long sequenceNumber,
         final long nativeHandle) {
       sequenceNumber_ = sequenceNumber;
-      writeBatch_ = new WriteBatch(nativeHandle);
+      writeBatch_ = new WriteBatch(nativeHandle, true);
     }
 
     /**

java/src/main/java/org/rocksdb/WriteBatch.java

@@ -59,8 +59,20 @@ public class WriteBatch extends AbstractWriteBatch {
    * @param nativeHandle address of native instance.
    */
   WriteBatch(final long nativeHandle) {
+    this(nativeHandle, false);
+  }
+
+  /**
+   * <p>Private WriteBatch constructor which is used to construct
+   * WriteBatch instances. </p>
+   *
+   * @param nativeHandle address of native instance.
+   * @param owningNativeHandle whether to own this reference from the C++ side or not
+   */
+  WriteBatch(final long nativeHandle, final boolean owningNativeHandle) {
     super(nativeHandle);
-    disOwnNativeHandle();
+    if(!owningNativeHandle)
+      disOwnNativeHandle();
   }
 
   @Override protected final native void disposeInternal(final long handle);