[arrow] branch master updated: ARROW-4610: [Plasma] Avoid Crash in Plasma Java Client

[ap...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

apitrou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git


The following commit(s) were added to refs/heads/master by this push:
     new b228489  ARROW-4610: [Plasma] Avoid Crash in Plasma Java Client
b228489 is described below

commit b22848952f09d6f9487feaff80ee358ca41b1562
Author: Yuhong Guo <yu...@antfin.com>
AuthorDate: Wed Feb 20 16:45:39 2019 +0100

    ARROW-4610: [Plasma] Avoid Crash in Plasma Java Client
    
    This PR removes all `ARROW_CHECK` from the JNI code to avoid Java client from crashing. The Java client will throw exception instead. For one thing, it is better to throw exceptions from the lower part and let the upper user to decide how to handle it. For another, JVM should not crash at all times and there are a lot of JVM core dump files when we are doing some failover tests by killing Plasma Server.
    
    Author: Yuhong Guo <yu...@antfin.com>
    
    Closes #3682 from guoyuhong/removeCheckFromJNI and squashes the following commits:
    
    3cbb4e3d9 <Yuhong Guo> Add Plasma Client Java test
    52c52bdd3 <Yuhong Guo> Avoid Crash in Plasma Java Client
---
 .../org_apache_arrow_plasma_PlasmaClientJNI.cc     | 28 ++++++++++++++-------
 .../plasma/exceptions/PlasmaClientException.java   | 29 ++++++++++++++++++++++
 .../org/apache/arrow/plasma/PlasmaClientTest.java  | 28 ++++++++++++++++++++-
 3 files changed, 75 insertions(+), 10 deletions(-)

diff --git a/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc b/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
index 1988742..248c268 100644
--- a/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
+++ b/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
@@ -42,6 +42,14 @@ inline void object_id_to_jbyteArray(JNIEnv* env, jbyteArray a, plasma::ObjectID*
   env->SetByteArrayRegion(a, 0, OBJECT_ID_SIZE, reinterpret_cast<jbyte*>(oid));
 }
 
+inline void throw_exception_if_not_OK(JNIEnv* env, const arrow::Status& status) {
+  if (!status.ok()) {
+    jclass Exception =
+        env->FindClass("org/apache/arrow/plasma/exceptions/PlasmaClientException");
+    env->ThrowNew(Exception, status.message().c_str());
+  }
+}
+
 class JByteArrayGetter {
  private:
   JNIEnv* _env;
@@ -67,7 +75,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_connect(
   const char* m_name = env->GetStringUTFChars(manager_socket_name, nullptr);
 
   plasma::PlasmaClient* client = new plasma::PlasmaClient();
-  ARROW_CHECK_OK(client->Connect(s_name, m_name, release_delay));
+  throw_exception_if_not_OK(env, client->Connect(s_name, m_name, release_delay));
 
   env->ReleaseStringUTFChars(store_socket_name, s_name);
   env->ReleaseStringUTFChars(manager_socket_name, m_name);
@@ -78,7 +86,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_disconnect(
     JNIEnv* env, jclass cls, jlong conn) {
   plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
 
-  ARROW_CHECK_OK(client->Disconnect());
+  throw_exception_if_not_OK(env, client->Disconnect());
   delete client;
   return;
 }
@@ -115,7 +123,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create(
     env->ThrowNew(exceptionClass, "");
     return nullptr;
   }
-  ARROW_CHECK(s.ok());
+  throw_exception_if_not_OK(env, s);
 
   return env->NewDirectByteBuffer(data->mutable_data(), size);
 }
@@ -145,7 +153,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_seal(
   plasma::ObjectID oid;
   jbyteArray_to_object_id(env, object_id, &oid);
 
-  ARROW_CHECK_OK(client->Seal(oid));
+  throw_exception_if_not_OK(env, client->Seal(oid));
 }
 
 JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release(
@@ -154,7 +162,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release(
   plasma::ObjectID oid;
   jbyteArray_to_object_id(env, object_id, &oid);
 
-  ARROW_CHECK_OK(client->Release(oid));
+  throw_exception_if_not_OK(env, client->Release(oid));
 }
 
 JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_delete(
@@ -163,7 +171,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_delete(
   plasma::ObjectID oid;
   jbyteArray_to_object_id(env, object_id, &oid);
 
-  ARROW_CHECK_OK(client->Delete(oid));
+  throw_exception_if_not_OK(env, client->Delete(oid));
 }
 
 JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get(
@@ -179,7 +187,8 @@ JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get(
         &oids[i]);
   }
   // TODO: may be blocked. consider to add the thread support
-  ARROW_CHECK_OK(client->Get(oids.data(), num_oids, timeout_ms, obufs.data()));
+  throw_exception_if_not_OK(env,
+                            client->Get(oids.data(), num_oids, timeout_ms, obufs.data()));
 
   jclass clsByteBuffer = env->FindClass("java/nio/ByteBuffer");
   jclass clsByteBufferArray = env->FindClass("[Ljava/nio/ByteBuffer;");
@@ -217,7 +226,7 @@ JNIEXPORT jboolean JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_contains
   jbyteArray_to_object_id(env, object_id, &oid);
 
   bool has_object;
-  ARROW_CHECK_OK(client->Contains(oid, &has_object));
+  throw_exception_if_not_OK(env, client->Contains(oid, &has_object));
 
   return has_object;
 }
@@ -227,7 +236,8 @@ JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_evict(
   plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
 
   int64_t evicted_bytes;
-  ARROW_CHECK_OK(client->Evict(static_cast<int64_t>(num_bytes), evicted_bytes));
+  throw_exception_if_not_OK(
+      env, client->Evict(static_cast<int64_t>(num_bytes), evicted_bytes));
 
   return static_cast<jlong>(evicted_bytes);
 }
diff --git a/java/plasma/src/main/java/org/apache/arrow/plasma/exceptions/PlasmaClientException.java b/java/plasma/src/main/java/org/apache/arrow/plasma/exceptions/PlasmaClientException.java
new file mode 100644
index 0000000..0608ad7
--- /dev/null
+++ b/java/plasma/src/main/java/org/apache/arrow/plasma/exceptions/PlasmaClientException.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.plasma.exceptions;
+
+public class PlasmaClientException extends RuntimeException {
+
+  public PlasmaClientException(String message) {
+    super(message);
+  }
+
+  public PlasmaClientException(String message, Throwable t) {
+    super(message, t);
+  }
+}
diff --git a/java/plasma/src/test/java/org/apache/arrow/plasma/PlasmaClientTest.java b/java/plasma/src/test/java/org/apache/arrow/plasma/PlasmaClientTest.java
index 3f326d3..e978a24 100644
--- a/java/plasma/src/test/java/org/apache/arrow/plasma/PlasmaClientTest.java
+++ b/java/plasma/src/test/java/org/apache/arrow/plasma/PlasmaClientTest.java
@@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import org.apache.arrow.plasma.exceptions.DuplicateObjectException;
+import org.apache.arrow.plasma.exceptions.PlasmaClientException;
 import org.junit.Assert;
 
 public class PlasmaClientTest {
@@ -205,7 +206,32 @@ public class PlasmaClientTest {
     assert !pLink.contains(id6);
     System.out.println("Plasma java client delete test success.");
     
-    cleanup();
+    // Test calling shuntdown while getting the object.
+    Thread thread = new Thread(() -> {
+      try {
+        TimeUnit.SECONDS.sleep(1);
+        cleanup();
+      } catch (InterruptedException e) {
+        throw new RuntimeException("Got InterruptedException when sleeping.", e);
+      }
+    });
+    thread.start();
+
+    try {
+      byte[] idNone =  new byte[20];
+      Arrays.fill(idNone, (byte)987);
+      pLink.get(idNone, timeoutMs, false);
+      Assert.fail("Fail to throw PlasmaClientException when get an object " +
+                  "when object store shutdown.");
+    } catch (PlasmaClientException e) {
+      System.out.println(String.format("Expected PlasmaClientException: %s", e));
+    }
+
+    try {
+      thread.join();
+    } catch (Exception e) {
+      System.out.println(String.format("Excpetion caught: %s", e));
+    }
     System.out.println("All test success.");
 
   }