From af6ec4d78e82985a065c8bc11e669c7e9aedfd03 Mon Sep 17 00:00:00 2001
From: Min Wei <min.mwei@gmail.cm>
Date: Fri, 20 Jan 2017 13:12:24 -0800
Subject: [PATCH] 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
---
 .../main/java/org/rocksdb/AbstractWriteBatch.java  | 11 -----------
 .../java/org/rocksdb/TransactionLogIterator.java   |  2 +-
 java/src/main/java/org/rocksdb/WriteBatch.java     | 14 +++++++++++++-
 3 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/java/src/main/java/org/rocksdb/AbstractWriteBatch.java b/java/src/main/java/org/rocksdb/AbstractWriteBatch.java
index cad7ebbd3..af3a249fd 100644
--- a/java/src/main/java/org/rocksdb/AbstractWriteBatch.java
+++ b/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_);
   }
 
diff --git a/java/src/main/java/org/rocksdb/TransactionLogIterator.java b/java/src/main/java/org/rocksdb/TransactionLogIterator.java
index f9684bd72..868548085 100644
--- a/java/src/main/java/org/rocksdb/TransactionLogIterator.java
+++ b/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);
     }
 
     /**
diff --git a/java/src/main/java/org/rocksdb/WriteBatch.java b/java/src/main/java/org/rocksdb/WriteBatch.java
index 325f9c057..bfd05a93f 100644
--- a/java/src/main/java/org/rocksdb/WriteBatch.java
+++ b/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);