[LIVY-892] Enhance Livy Java Client to execute statements [WIP]

api/src/main/java/org/apache/livy/JobHandle.java

@@ -17,7 +17,6 @@
 
 package org.apache.livy;
 
-import java.util.List;
 import java.util.concurrent.Future;
 
 /**
@@ -32,6 +31,13 @@ public interface JobHandle<T> extends Future<T> {
    */
   State getState();
 
+  /**
+   * Return the current state of the job.
+   *
+   * @return The current State of this job
+   */
+  String getId();
+
   /**
    * Add a listener to the job handle. If the job's state is not SENT, a callback for the
    * corresponding state will be invoked immediately.
@@ -47,6 +53,7 @@ static enum State {
     SENT,
     QUEUED,
     STARTED,
+    CANCELLING,
     CANCELLED,
     FAILED,
     SUCCEEDED;

api/src/main/java/org/apache/livy/LivyClient.java

@@ -26,6 +26,15 @@
  */
 public interface LivyClient {
 
+  /**
+   * Submits a code snippet passed in as a string for asynchronous execution.
+   * @param <T> The return type of the statement
+   * @param code The code to execute
+   * @param kind The code kind - can be one of (spark, pyspark sparkr, sql)
+   * @return A handle that can be used monitor the statement execution.
+   */
+  <T> JobHandle<T> submitStatement(String code, String kind);
+
   /**
    * Submits a job for asynchronous execution.
    *

api/src/test/java/org/apache/livy/TestClientFactory.java

@@ -56,6 +56,11 @@ private Client(Properties config) {
       this.config = config;
     }
 
+    @Override
+    public <T> JobHandle<T> submitStatement(String code, String kind){
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public <T> JobHandle<T> submit(Job<T> job) {
       throw new UnsupportedOperationException();

client-common/src/main/java/org/apache/livy/client/common/AbstractJobHandle.java

@@ -17,13 +17,12 @@
 
 package org.apache.livy.client.common;
 
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-
 import org.apache.livy.JobHandle;
 import org.apache.livy.annotations.Private;
 
+import java.util.LinkedList;
+import java.util.List;
+
 @Private
 public abstract class AbstractJobHandle<T> implements JobHandle<T> {
 
@@ -35,6 +34,10 @@ protected AbstractJobHandle() {
     this.state = State.SENT;
   }
 
+  public String getId(){
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public JobHandle.State getState() {
     return state;

client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java

@@ -17,12 +17,12 @@
 
 package org.apache.livy.client.common;
 
-import java.util.List;
-import java.util.Map;
-
 import org.apache.livy.JobHandle.State;
 import org.apache.livy.annotations.Private;
 
+import java.util.List;
+import java.util.Map;
+
 /**
  * There are the Java representations of the JSON messages used by the client protocol.
  *
@@ -81,6 +81,22 @@ private SessionInfo() {
 
   }
 
+  public static class SerializedStatement implements ClientMessage {
+
+    public final String code;
+    public final String kind;
+
+    public SerializedStatement(String code, String kind) {
+      this.code = code;
+      this.kind = kind;
+    }
+
+    private SerializedStatement() {
+      this(null, null);
+    }
+
+  }
+
   public static class SerializedJob implements ClientMessage {
 
     public final byte[] job;

client-http/src/main/java/org/apache/livy/client/http/HttpClient.java

@@ -17,23 +17,20 @@
 
 package org.apache.livy.client.http;
 
+import org.apache.livy.Job;
+import org.apache.livy.JobHandle;
+import org.apache.livy.LivyClient;
+import org.apache.livy.client.common.Serializer;
+
 import java.io.File;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.apache.livy.Job;
-import org.apache.livy.JobHandle;
-import org.apache.livy.LivyClient;
-import org.apache.livy.client.common.Serializer;
 import static org.apache.livy.client.common.HttpMessages.*;
 
 /**
@@ -96,6 +93,13 @@ public Thread newThread(Runnable r) {
     this.serializer = new Serializer();
   }
 
+  @Override
+  public <T> JobHandle<T> submitStatement(String code, String kind) {
+    StatementHandleImpl<T> handle = new StatementHandleImpl<T>(config, conn, sessionId, executor, serializer);
+    handle.start("statements", code, kind);
+    return handle;
+  }
+
   @Override
   public <T> JobHandle<T> submit(Job<T> job) {
     return sendJob("submit-job", job);

client-http/src/main/java/org/apache/livy/client/http/JobHandleImpl.java

@@ -17,17 +17,13 @@
 
 package org.apache.livy.client.http;
 
-import java.nio.ByteBuffer;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
 import org.apache.livy.client.common.AbstractJobHandle;
 import org.apache.livy.client.common.BufferUtils;
 import org.apache.livy.client.common.Serializer;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.*;
+
 import static org.apache.livy.client.common.HttpMessages.*;
 
 class JobHandleImpl<T> extends AbstractJobHandle<T> {
@@ -133,6 +129,10 @@ protected Throwable error() {
     return error;
   }
 
+  public String getId() {
+    return Long.toString(jobId);
+  }
+
   void start(final String command, final ByteBuffer serializedJob) {
     Runnable task = new Runnable() {
       @Override