You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@livy.apache.org by js...@apache.org on 2017/07/05 03:07:00 UTC
[31/33] incubator-livy git commit: LIVY-375. Change Livy code package
name to org.apache.livy
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-common/src/test/java/org/apache/livy/client/common/TestHttpMessages.java
----------------------------------------------------------------------
diff --git a/client-common/src/test/java/org/apache/livy/client/common/TestHttpMessages.java b/client-common/src/test/java/org/apache/livy/client/common/TestHttpMessages.java
new file mode 100644
index 0000000..53d4107
--- /dev/null
+++ b/client-common/src/test/java/org/apache/livy/client/common/TestHttpMessages.java
@@ -0,0 +1,130 @@
+/*
+ * 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.livy.client.common;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import org.apache.livy.JobHandle.State;
+
+public class TestHttpMessages {
+
+ /**
+ * Tests that all defined messages can be serialized and deserialized using Jackson.
+ */
+ @Test
+ public void testMessageSerialization() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+
+ for (Class<?> msg : HttpMessages.class.getClasses()) {
+ if (msg.isInterface()) {
+ continue;
+ }
+
+ String name = msg.getSimpleName();
+
+ Constructor c = msg.getConstructors()[0];
+ Object[] params = new Object[c.getParameterTypes().length];
+ Type[] genericTypes = c.getGenericParameterTypes();
+ for (int i = 0; i < params.length; i++) {
+ params[i] = dummyValue(c.getParameterTypes()[i], genericTypes[i]);
+ }
+
+ Object o1 = c.newInstance(params);
+ byte[] serialized = mapper.writeValueAsBytes(o1);
+ Object o2 = mapper.readValue(serialized, msg);
+
+ assertNotNull("could not deserialize " + name, o2);
+ for (Field f : msg.getFields()) {
+ checkEquals(name, f, o1, o2);
+ }
+ }
+
+ }
+
+ @Test(expected=IllegalArgumentException.class)
+ public void testJobStatusResultBadState() {
+ new HttpMessages.JobStatus(0L, State.QUEUED, new byte[1], null);
+ }
+
+ @Test(expected=IllegalArgumentException.class)
+ public void testJobStatusErrorBadState() {
+ new HttpMessages.JobStatus(0L, State.QUEUED, null, "An Error");
+ }
+
+ private Object dummyValue(Class<?> klass, Type type) {
+ switch (klass.getSimpleName()) {
+ case "int":
+ case "Integer":
+ return 42;
+ case "long": return 84L;
+ case "byte[]": return new byte[] { (byte) 0x42, (byte) 0x84 };
+ case "String": return "test";
+ case "State": return State.SUCCEEDED;
+ case "Map":
+ Map<String, String> map = new HashMap<>();
+ map.put("dummy1", "dummy2");
+ return map;
+ case "List":
+ Class<?> genericType = getGenericArgType(type);
+ return Arrays.asList(dummyValue(genericType, null), dummyValue(genericType, null));
+ default: throw new IllegalArgumentException("FIX ME: " + klass.getSimpleName());
+ }
+ }
+
+ private Class<?> getGenericArgType(Type type) {
+ assertNotNull("FIX ME: null type argument.", type);
+
+ ParameterizedType ptype = (ParameterizedType) type;
+ assertEquals("FIX ME: no support for multiple type arguments.",
+ 1, ptype.getActualTypeArguments().length);
+
+ Type argType = ptype.getActualTypeArguments()[0];
+ assertTrue("FIX ME: type argument is not a class.", argType instanceof Class);
+
+ return (Class<?>) argType;
+ }
+
+ private void checkEquals(String name, Field f, Object o1, Object o2) throws Exception {
+ Object v1 = f.get(o1);
+ Object v2 = f.get(o2);
+
+ boolean match;
+ if (!f.getType().isArray()) {
+ match = v1.equals(v2);
+ } else if (v1 instanceof byte[]) {
+ match = Arrays.equals((byte[]) v1, (byte[]) v2);
+ } else {
+ throw new IllegalArgumentException("FIX ME: " + f.getType().getSimpleName());
+ }
+
+ assertTrue(
+ String.format("Field %s of %s does not match after deserialization.", f.getName(), name),
+ match);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-common/src/test/java/org/apache/livy/client/common/TestSerializer.java
----------------------------------------------------------------------
diff --git a/client-common/src/test/java/org/apache/livy/client/common/TestSerializer.java b/client-common/src/test/java/org/apache/livy/client/common/TestSerializer.java
new file mode 100644
index 0000000..65fa555
--- /dev/null
+++ b/client-common/src/test/java/org/apache/livy/client/common/TestSerializer.java
@@ -0,0 +1,78 @@
+/*
+ * 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.livy.client.common;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class TestSerializer {
+
+ private static final String MESSAGE = "Hello World!";
+
+ @Test
+ public void testSerializer() throws Exception {
+ Object decoded = doSerDe(MESSAGE);
+ assertEquals(MESSAGE, decoded);
+ }
+
+ @Test
+ public void testUnicodeSerializer() throws Exception {
+ StringBuilder builder = new StringBuilder();
+ for (int x = 0; x < 5000; x++) {
+ builder.append("\u263A");
+ }
+ String testMessage = builder.toString();
+ Object decoded = doSerDe(testMessage);
+ assertEquals(testMessage, decoded);
+ }
+
+ @Test
+ public void testAutoRegistration() throws Exception {
+ Object decoded = doSerDe(new TestMessage(MESSAGE), TestMessage.class);
+ assertTrue(decoded instanceof TestMessage);
+ assertEquals(MESSAGE, ((TestMessage)decoded).data);
+ }
+
+ private Object doSerDe(Object data, Class<?>... klasses) {
+ Serializer s = new Serializer(klasses);
+ ByteBuffer serialized = s.serialize(data);
+ return s.deserialize(serialized);
+ }
+
+ private ByteBuffer newBuffer() {
+ return ByteBuffer.allocate(1024);
+ }
+
+ private static class TestMessage {
+ final String data;
+
+ TestMessage() {
+ this(null);
+ }
+
+ TestMessage(String data) {
+ this.data = data;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-common/src/test/java/org/apache/livy/client/common/TestTestUtils.java
----------------------------------------------------------------------
diff --git a/client-common/src/test/java/org/apache/livy/client/common/TestTestUtils.java b/client-common/src/test/java/org/apache/livy/client/common/TestTestUtils.java
new file mode 100644
index 0000000..bffb72f
--- /dev/null
+++ b/client-common/src/test/java/org/apache/livy/client/common/TestTestUtils.java
@@ -0,0 +1,38 @@
+/*
+ * 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.livy.client.common;
+
+import java.nio.ByteBuffer;
+
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class TestTestUtils {
+
+ @Test
+ public void testJacocoArgs() {
+ String args1 = TestUtils.getJacocoArgs();
+ String expected1 = System.getProperty("jacoco.args").replace("main.exec", "jacoco-1.exec");
+ assertEquals(expected1, args1);
+
+ String args2 = TestUtils.getJacocoArgs();
+ String expected2 = System.getProperty("jacoco.args").replace("main.exec", "jacoco-2.exec");
+ assertEquals(expected2, args2);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/pom.xml
----------------------------------------------------------------------
diff --git a/client-http/pom.xml b/client-http/pom.xml
index 056d4c0..15e16db 100644
--- a/client-http/pom.xml
+++ b/client-http/pom.xml
@@ -18,37 +18,37 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-main</artifactId>
- <version>0.4.0-SNAPSHOT</version>
+ <version>0.4.0-incubating-SNAPSHOT</version>
</parent>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-client-http</artifactId>
- <version>0.4.0-SNAPSHOT</version>
+ <version>0.4.0-incubating-SNAPSHOT</version>
<packaging>jar</packaging>
<dependencies>
<dependency>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-client-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-test-lib</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-core_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
@@ -80,7 +80,7 @@
</dependency>
<dependency>
- <groupId>com.cloudera.livy</groupId>
+ <groupId>org.apache.livy</groupId>
<artifactId>livy-server</artifactId>
<version>${project.version}</version>
<scope>test</scope>
@@ -126,7 +126,7 @@
<include>*:*</include>
</includes>
<excludes>
- <exclude>com.cloudera.livy:livy-api</exclude>
+ <exclude>org.apache.livy:livy-api</exclude>
</excludes>
</artifactSet>
<filters>
@@ -142,19 +142,19 @@
<relocations>
<relocation>
<pattern>com.esotericsoftware</pattern>
- <shadedPattern>com.cloudera.livy.shaded.kryo</shadedPattern>
+ <shadedPattern>org.apache.livy.shaded.kryo</shadedPattern>
</relocation>
<relocation>
<pattern>com.fasterxml.jackson</pattern>
- <shadedPattern>com.cloudera.livy.shaded.jackson</shadedPattern>
+ <shadedPattern>org.apache.livy.shaded.jackson</shadedPattern>
</relocation>
<relocation>
<pattern>org.apache.commons</pattern>
- <shadedPattern>com.cloudera.livy.shaded.apache.commons</shadedPattern>
+ <shadedPattern>org.apache.livy.shaded.apache.commons</shadedPattern>
</relocation>
<relocation>
<pattern>org.apache.http</pattern>
- <shadedPattern>com.cloudera.livy.shaded.apache.http</shadedPattern>
+ <shadedPattern>org.apache.livy.shaded.apache.http</shadedPattern>
</relocation>
</relocations>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/com/cloudera/livy/client/http/HttpClient.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/com/cloudera/livy/client/http/HttpClient.java b/client-http/src/main/java/com/cloudera/livy/client/http/HttpClient.java
deleted file mode 100644
index 943fdb5..0000000
--- a/client-http/src/main/java/com/cloudera/livy/client/http/HttpClient.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/*
- * 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 com.cloudera.livy.client.http;
-
-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.regex.Matcher;
-import java.util.regex.Pattern;
-
-import com.cloudera.livy.Job;
-import com.cloudera.livy.JobHandle;
-import com.cloudera.livy.LivyClient;
-import com.cloudera.livy.client.common.BufferUtils;
-import com.cloudera.livy.client.common.Serializer;
-import static com.cloudera.livy.client.common.HttpMessages.*;
-import static com.cloudera.livy.client.http.HttpConf.Entry.*;
-
-/**
- * What is currently missing:
- * - monitoring of spark job IDs launched by jobs
- */
-class HttpClient implements LivyClient {
-
- private final HttpConf config;
- private final LivyConnection conn;
- private final int sessionId;
- private final ScheduledExecutorService executor;
- private final Serializer serializer;
-
- private boolean stopped;
-
- HttpClient(URI uri, HttpConf httpConf) {
- this.config = httpConf;
- this.stopped = false;
-
- // If the given URI looks like it refers to an existing session, then try to connect to
- // an existing session. Note this means that any Spark configuration in httpConf will be
- // unused.
- Matcher m = Pattern.compile("(.*)" + LivyConnection.SESSIONS_URI + "/([0-9]+)")
- .matcher(uri.getPath());
-
- try {
- if (m.matches()) {
- URI base = new URI(uri.getScheme(), uri.getUserInfo(), uri.getHost(), uri.getPort(),
- m.group(1), uri.getQuery(), uri.getFragment());
-
- this.conn = new LivyConnection(base, httpConf);
- this.sessionId = Integer.parseInt(m.group(2));
- conn.post(null, SessionInfo.class, "/%d/connect", sessionId);
- } else {
- Map<String, String> sessionConf = new HashMap<>();
- for (Map.Entry<String, String> e : config) {
- sessionConf.put(e.getKey(), e.getValue());
- }
-
- ClientMessage create = new CreateClientRequest(sessionConf);
- this.conn = new LivyConnection(uri, httpConf);
- this.sessionId = conn.post(create, SessionInfo.class, "/").id;
- }
- } catch (Exception e) {
- throw propagate(e);
- }
-
- // Because we only have one connection to the server, we don't need more than a single
- // threaded executor here.
- this.executor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "HttpClient-" + sessionId);
- t.setDaemon(true);
- return t;
- }
- });
-
- this.serializer = new Serializer();
- }
-
- @Override
- public <T> JobHandle<T> submit(Job<T> job) {
- return sendJob("submit-job", job);
- }
-
- @Override
- public <T> Future<T> run(Job<T> job) {
- return sendJob("run-job", job);
- }
-
- @Override
- public synchronized void stop(boolean shutdownContext) {
- if (!stopped) {
- executor.shutdownNow();
- try {
- if (shutdownContext) {
- conn.delete(Map.class, "/%s", sessionId);
- }
- } catch (Exception e) {
- throw propagate(e);
- } finally {
- try {
- conn.close();
- } catch (Exception e) {
- // Ignore.
- }
- }
- stopped = true;
- }
- }
-
- public Future<?> uploadJar(File jar) {
- return uploadResource(jar, "upload-jar", "jar");
- }
-
- @Override
- public Future<?> addJar(URI uri) {
- return addResource("add-jar", uri);
- }
-
- public Future<?> uploadFile(File file) {
- return uploadResource(file, "upload-file", "file");
- }
-
- @Override
- public Future<?> addFile(URI uri) {
- return addResource("add-file", uri);
- }
-
- private Future<?> uploadResource(final File file, final String command, final String paramName) {
- Callable<Void> task = new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- conn.post(file, Void.class, paramName, "/%d/%s", sessionId, command);
- return null;
- }
- };
- return executor.submit(task);
- }
-
- private Future<?> addResource(final String command, final URI resource) {
- Callable<Void> task = new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- ClientMessage msg = new AddResource(resource.toString());
- conn.post(msg, Void.class, "/%d/%s", sessionId, command);
- return null;
- }
- };
- return executor.submit(task);
- }
-
- private <T> JobHandleImpl<T> sendJob(final String command, Job<T> job) {
- final ByteBuffer serializedJob = serializer.serialize(job);
- JobHandleImpl<T> handle = new JobHandleImpl<T>(config, conn, sessionId, executor, serializer);
- handle.start(command, serializedJob);
- return handle;
- }
-
- private RuntimeException propagate(Exception cause) {
- if (cause instanceof RuntimeException) {
- throw (RuntimeException) cause;
- } else {
- throw new RuntimeException(cause);
- }
- }
-
- // For testing.
- int getSessionId() {
- return sessionId;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/com/cloudera/livy/client/http/HttpClientFactory.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/com/cloudera/livy/client/http/HttpClientFactory.java b/client-http/src/main/java/com/cloudera/livy/client/http/HttpClientFactory.java
deleted file mode 100644
index 2062ef2..0000000
--- a/client-http/src/main/java/com/cloudera/livy/client/http/HttpClientFactory.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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 com.cloudera.livy.client.http;
-
-import java.net.URI;
-import java.util.Properties;
-
-import com.cloudera.livy.LivyClient;
-import com.cloudera.livy.LivyClientFactory;
-
-/**
- * Factory for HTTP Livy clients.
- */
-public final class HttpClientFactory implements LivyClientFactory {
-
- @Override
- public LivyClient createClient(URI uri, Properties config) {
- if (!"http".equals(uri.getScheme()) && !"https".equals(uri.getScheme())) {
- return null;
- }
-
- return new HttpClient(uri, new HttpConf(config));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/com/cloudera/livy/client/http/HttpConf.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/com/cloudera/livy/client/http/HttpConf.java b/client-http/src/main/java/com/cloudera/livy/client/http/HttpConf.java
deleted file mode 100644
index 4ca3665..0000000
--- a/client-http/src/main/java/com/cloudera/livy/client/http/HttpConf.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * 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 com.cloudera.livy.client.http;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-
-import com.cloudera.livy.client.common.ClientConf;
-
-class HttpConf extends ClientConf<HttpConf> {
-
- private static final String HTTP_CONF_PREFIX = "livy.client.http.";
-
- static enum Entry implements ConfEntry {
- CONNECTION_TIMEOUT("connection.timeout", "10s"),
- CONNECTION_IDLE_TIMEOUT("connection.idle.timeout", "10m"),
- SOCKET_TIMEOUT("connection.socket.timeout", "5m"),
-
- JOB_INITIAL_POLL_INTERVAL("job.initial-poll-interval", "100ms"),
- JOB_MAX_POLL_INTERVAL("job.max-poll-interval", "5s"),
-
- CONTENT_COMPRESS_ENABLE("content.compress.enable", true),
-
- // Kerberos related configuration
- SPNEGO_ENABLED("spnego.enable", false),
- AUTH_LOGIN_CONFIG("auth.login.config", null),
- KRB5_DEBUG_ENABLED("krb5.debug", false),
- KRB5_CONF("krb5.conf", null);
-
- private final String key;
- private final Object dflt;
-
- private Entry(String key, Object dflt) {
- this.key = HTTP_CONF_PREFIX + key;
- this.dflt = dflt;
- }
-
- @Override
- public String key() { return key; }
-
- @Override
- public Object dflt() { return dflt; }
- }
-
- HttpConf(Properties config) {
- super(config);
-
- if (getBoolean(Entry.SPNEGO_ENABLED)) {
- if (get(Entry.AUTH_LOGIN_CONFIG ) == null) {
- throw new IllegalArgumentException(Entry.AUTH_LOGIN_CONFIG.key + " should not be null");
- }
-
- if (get(Entry.KRB5_CONF) == null) {
- throw new IllegalArgumentException(Entry.KRB5_CONF.key + " should not be null");
- }
-
- System.setProperty("java.security.auth.login.config", get(Entry.AUTH_LOGIN_CONFIG));
- System.setProperty("java.security.krb5.conf", get(Entry.KRB5_CONF));
- System.setProperty(
- "sun.security.krb5.debug", String.valueOf(getBoolean(Entry.KRB5_DEBUG_ENABLED)));
- // This is needed to get Kerberos credentials from the environment, instead of
- // requiring the application to manually obtain the credentials.
- System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
- }
- }
-
- boolean isSpnegoEnabled() {
- return getBoolean(Entry.SPNEGO_ENABLED);
- }
-
- private static final Map<String, DeprecatedConf> configsWithAlternatives
- = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
- put(HttpConf.Entry.JOB_INITIAL_POLL_INTERVAL.key, DepConf.JOB_INITIAL_POLL_INTERVAL);
- put(HttpConf.Entry.JOB_MAX_POLL_INTERVAL.key, DepConf.JOB_MAX_POLL_INTERVAL);
- }});
-
- // Maps deprecated key to DeprecatedConf with the same key.
- // There are no deprecated configs without alternatives currently.
- private static final Map<String, DeprecatedConf> deprecatedConfigs
- = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>());
-
- protected Map<String, DeprecatedConf> getConfigsWithAlternatives() {
- return configsWithAlternatives;
- }
-
- protected Map<String, DeprecatedConf> getDeprecatedConfigs() {
- return deprecatedConfigs;
- }
-
- static enum DepConf implements DeprecatedConf {
- JOB_INITIAL_POLL_INTERVAL("job.initial_poll_interval", "0.4"),
- JOB_MAX_POLL_INTERVAL("job.max_poll_interval", "0.4");
-
- private final String key;
- private final String version;
- private final String deprecationMessage;
-
- private DepConf(String key, String version) {
- this(key, version, "");
- }
-
- private DepConf(String key, String version, String deprecationMessage) {
- this.key = HTTP_CONF_PREFIX + key;
- this.version = version;
- this.deprecationMessage = deprecationMessage;
- }
-
- @Override
- public String key() { return key; }
-
- @Override
- public String version() { return version; }
-
- @Override
- public String deprecationMessage() { return deprecationMessage; }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/com/cloudera/livy/client/http/JobHandleImpl.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/com/cloudera/livy/client/http/JobHandleImpl.java b/client-http/src/main/java/com/cloudera/livy/client/http/JobHandleImpl.java
deleted file mode 100644
index e437059..0000000
--- a/client-http/src/main/java/com/cloudera/livy/client/http/JobHandleImpl.java
+++ /dev/null
@@ -1,275 +0,0 @@
-/*
- * 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 com.cloudera.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 com.cloudera.livy.JobHandle;
-import com.cloudera.livy.client.common.AbstractJobHandle;
-import com.cloudera.livy.client.common.BufferUtils;
-import com.cloudera.livy.client.common.Serializer;
-import static com.cloudera.livy.client.common.HttpMessages.*;
-import static com.cloudera.livy.client.http.HttpConf.Entry.*;
-
-class JobHandleImpl<T> extends AbstractJobHandle<T> {
-
- private final long sessionId;
- private final LivyConnection conn;
- private final ScheduledExecutorService executor;
- private final Object lock;
- private final Serializer serializer;
-
- private final long initialPollInterval;
- private final long maxPollInterval;
-
- private long jobId;
- private T result;
- private Throwable error;
- private volatile boolean isDone;
- private volatile boolean isCancelled;
- private volatile boolean isCancelPending;
- private volatile ScheduledFuture<?> pollTask;
-
- JobHandleImpl(
- HttpConf config,
- LivyConnection conn,
- long sessionId,
- ScheduledExecutorService executor,
- Serializer s) {
- this.conn = conn;
- this.sessionId = sessionId;
- this.executor = executor;
- this.lock = new Object();
- this.serializer = s;
- this.isDone = false;
-
- this.initialPollInterval = config.getTimeAsMs(JOB_INITIAL_POLL_INTERVAL);
- this.maxPollInterval = config.getTimeAsMs(JOB_MAX_POLL_INTERVAL);
-
- if (initialPollInterval <= 0) {
- throw new IllegalArgumentException("Invalid initial poll interval.");
- }
- if (maxPollInterval <= 0 || maxPollInterval < initialPollInterval) {
- throw new IllegalArgumentException(
- "Invalid max poll interval, or lower than initial interval.");
- }
-
- // The job ID is set asynchronously, and there might be a call to cancel() before it's
- // set. So cancel() will always set the isCancelPending flag, even if there's no job
- // ID yet. If the thread setting the job ID sees that flag, it will send a cancel request
- // to the server. There's still a possibility that two cancel requests will be sent,
- // but that doesn't cause any harm.
- this.isCancelPending = false;
- this.jobId = -1;
- }
-
- @Override
- public T get() throws ExecutionException, InterruptedException {
- try {
- return get(true, -1, TimeUnit.MILLISECONDS);
- } catch (TimeoutException te) {
- // Not gonna happen.
- throw new RuntimeException(te);
- }
- }
-
- @Override
- public T get(long timeout, TimeUnit unit)
- throws ExecutionException, InterruptedException, TimeoutException {
- return get(false, timeout, unit);
- }
-
- @Override
- public boolean isDone() {
- return isDone;
- }
-
- @Override
- public boolean isCancelled() {
- return isCancelled;
- }
-
- @Override
- public boolean cancel(final boolean mayInterrupt) {
- // Do a best-effort to detect if already cancelled, but the final say is always
- // on the server side. Don't block the caller, though.
- if (!isCancelled && !isCancelPending) {
- isCancelPending = true;
- if (jobId > -1) {
- sendCancelRequest(jobId);
- }
- return true;
- }
-
- return false;
- }
-
- @Override
- protected T result() {
- return result;
- }
-
- @Override
- protected Throwable error() {
- return error;
- }
-
- void start(final String command, final ByteBuffer serializedJob) {
- Runnable task = new Runnable() {
- @Override
- public void run() {
- try {
- ClientMessage msg = new SerializedJob(BufferUtils.toByteArray(serializedJob));
- JobStatus status = conn.post(msg, JobStatus.class, "/%d/%s", sessionId, command);
-
- if (isCancelPending) {
- sendCancelRequest(status.id);
- }
-
- jobId = status.id;
-
- pollTask = executor.schedule(new JobPollTask(initialPollInterval),
- initialPollInterval, TimeUnit.MILLISECONDS);
- } catch (Exception e) {
- setResult(null, e, State.FAILED);
- }
- }
- };
- executor.submit(task);
- }
-
- private void sendCancelRequest(final long id) {
- executor.submit(new Runnable() {
- @Override
- public void run() {
- try {
- conn.post(null, Void.class, "/%d/jobs/%d/cancel", sessionId, id);
- } catch (Exception e) {
- setResult(null, e, State.FAILED);
- }
- }
- });
- }
-
- private T get(boolean waitIndefinitely, long timeout, TimeUnit unit)
- throws ExecutionException, InterruptedException, TimeoutException {
- if (!isDone) {
- synchronized (lock) {
- if (waitIndefinitely) {
- while (!isDone) {
- lock.wait();
- }
- } else {
- long now = System.nanoTime();
- long deadline = now + unit.toNanos(timeout);
- while (!isDone && deadline > now) {
- lock.wait(TimeUnit.NANOSECONDS.toMillis(deadline - now));
- now = System.nanoTime();
- }
- if (!isDone) {
- throw new TimeoutException();
- }
- }
- }
- }
- if (isCancelled) {
- throw new CancellationException();
- }
- if (error != null) {
- throw new ExecutionException(error);
- }
- return result;
- }
-
- private void setResult(T result, Throwable error, State newState) {
- if (!isDone) {
- synchronized (lock) {
- if (!isDone) {
- this.result = result;
- this.error = error;
- this.isDone = true;
- changeState(newState);
- }
- lock.notifyAll();
- }
- }
- }
-
- private class JobPollTask implements Runnable {
-
- private long currentInterval;
-
- JobPollTask(long currentInterval) {
- this.currentInterval = currentInterval;
- }
-
- @Override
- public void run() {
- try {
- JobStatus status = conn.get(JobStatus.class, "/%d/jobs/%d", sessionId, jobId);
- T result = null;
- Throwable error = null;
- boolean finished = false;
-
- switch (status.state) {
- case SUCCEEDED:
- if (status.result != null) {
- @SuppressWarnings("unchecked")
- T localResult = (T) serializer.deserialize(ByteBuffer.wrap(status.result));
- result = localResult;
- }
- finished = true;
- break;
-
- case FAILED:
- // TODO: better exception.
- error = new RuntimeException(status.error);
- finished = true;
- break;
-
- case CANCELLED:
- isCancelled = true;
- finished = true;
- break;
-
- default:
- // Nothing to do.
- }
- if (finished) {
- setResult(result, error, status.state);
- } else if (status.state != state) {
- changeState(status.state);
- }
- if (!finished) {
- currentInterval = Math.min(currentInterval * 2, maxPollInterval);
- pollTask = executor.schedule(this, currentInterval, TimeUnit.MILLISECONDS);
- }
- } catch (Exception e) {
- setResult(null, e, State.FAILED);
- }
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/com/cloudera/livy/client/http/LivyConnection.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/com/cloudera/livy/client/http/LivyConnection.java b/client-http/src/main/java/com/cloudera/livy/client/http/LivyConnection.java
deleted file mode 100644
index e48013f..0000000
--- a/client-http/src/main/java/com/cloudera/livy/client/http/LivyConnection.java
+++ /dev/null
@@ -1,235 +0,0 @@
-/*
- * 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 com.cloudera.livy.client.http;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URLDecoder;
-import java.nio.charset.StandardCharsets;
-import java.security.Principal;
-import java.util.concurrent.TimeUnit;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpHeaders;
-import org.apache.http.HttpStatus;
-import org.apache.http.auth.AuthSchemeProvider;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.Credentials;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.client.config.AuthSchemes;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.*;
-import org.apache.http.client.protocol.HttpClientContext;
-import org.apache.http.config.Registry;
-import org.apache.http.config.RegistryBuilder;
-import org.apache.http.entity.ByteArrayEntity;
-import org.apache.http.entity.mime.MultipartEntityBuilder;
-import org.apache.http.entity.mime.content.FileBody;
-import org.apache.http.impl.NoConnectionReuseStrategy;
-import org.apache.http.impl.auth.SPNegoSchemeFactory;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.conn.BasicHttpClientConnectionManager;
-import org.apache.http.util.EntityUtils;
-
-import static com.cloudera.livy.client.http.HttpConf.Entry.*;
-
-/**
- * Abstracts a connection to the Livy server; serializes multiple requests so that we only need
- * one active HTTP connection (to keep resource usage down).
- */
-class LivyConnection {
-
- static final String SESSIONS_URI = "/sessions";
- private static final String APPLICATION_JSON = "application/json";
-
- private final URI server;
- private final String uriRoot;
- private final CloseableHttpClient client;
- private final ObjectMapper mapper;
-
- LivyConnection(URI uri, final HttpConf config) {
- HttpClientContext ctx = HttpClientContext.create();
- int port = uri.getPort() > 0 ? uri.getPort() : 8998;
-
- String path = uri.getPath() != null ? uri.getPath() : "";
- this.uriRoot = path + SESSIONS_URI;
-
- RequestConfig reqConfig = new RequestConfig() {
- @Override
- public int getConnectTimeout() {
- return (int) config.getTimeAsMs(CONNECTION_TIMEOUT);
- }
-
- @Override
- public int getSocketTimeout() {
- return (int) config.getTimeAsMs(SOCKET_TIMEOUT);
- }
-
- @Override
- public boolean isAuthenticationEnabled() {
- return true;
- }
-
- @Override
- public boolean isContentCompressionEnabled() {
- return config.getBoolean(CONTENT_COMPRESS_ENABLE);
- }
- };
-
- Credentials credentials;
- // If user info is specified in the url, pass them to the CredentialsProvider.
- if (uri.getUserInfo() != null) {
- String[] userInfo = uri.getUserInfo().split(":");
- if (userInfo.length < 1) {
- throw new IllegalArgumentException("Malformed user info in the url.");
- }
- try {
- String username = URLDecoder.decode(userInfo[0], StandardCharsets.UTF_8.name());
- String password = "";
- if (userInfo.length > 1) {
- password = URLDecoder.decode(userInfo[1], StandardCharsets.UTF_8.name());
- }
- credentials = new UsernamePasswordCredentials(username, password);
- } catch (Exception e) {
- throw new IllegalArgumentException("User info in the url contains bad characters.", e);
- }
- } else {
- credentials = new Credentials() {
- @Override
- public String getPassword() {
- return null;
- }
-
- @Override
- public Principal getUserPrincipal() {
- return null;
- }
- };
- }
-
- CredentialsProvider credsProvider = new BasicCredentialsProvider();
- credsProvider.setCredentials(AuthScope.ANY, credentials);
-
- HttpClientBuilder builder = HttpClientBuilder.create()
- .disableAutomaticRetries()
- .evictExpiredConnections()
- .evictIdleConnections(config.getTimeAsMs(CONNECTION_IDLE_TIMEOUT), TimeUnit.MILLISECONDS)
- .setConnectionManager(new BasicHttpClientConnectionManager())
- .setConnectionReuseStrategy(new NoConnectionReuseStrategy())
- .setDefaultRequestConfig(reqConfig)
- .setMaxConnTotal(1)
- .setDefaultCredentialsProvider(credsProvider)
- .setUserAgent("livy-client-http");
-
- if (config.isSpnegoEnabled()) {
- Registry<AuthSchemeProvider> authSchemeProviderRegistry =
- RegistryBuilder.<AuthSchemeProvider>create()
- .register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory())
- .build();
- builder.setDefaultAuthSchemeRegistry(authSchemeProviderRegistry);
- }
-
- this.server = uri;
- this.client = builder.build();
- this.mapper = new ObjectMapper();
- }
-
- synchronized void close() throws IOException {
- client.close();
- }
-
- synchronized <V> V delete(Class<V> retType, String uri, Object... uriParams) throws Exception {
- return sendJSONRequest(new HttpDelete(), retType, uri, uriParams);
- }
-
- synchronized <V> V get(Class<V> retType, String uri, Object... uriParams) throws Exception {
- return sendJSONRequest(new HttpGet(), retType, uri, uriParams);
- }
-
- synchronized <V> V post(
- Object body,
- Class<V> retType,
- String uri,
- Object... uriParams) throws Exception {
- HttpPost post = new HttpPost();
- if (body != null) {
- byte[] bodyBytes = mapper.writeValueAsBytes(body);
- post.setEntity(new ByteArrayEntity(bodyBytes));
- }
- return sendJSONRequest(post, retType, uri, uriParams);
- }
-
- synchronized <V> V post(
- File f,
- Class<V> retType,
- String paramName,
- String uri,
- Object... uriParams) throws Exception {
- HttpPost post = new HttpPost();
- MultipartEntityBuilder builder = MultipartEntityBuilder.create();
- builder.addPart(paramName, new FileBody(f));
- post.setEntity(builder.build());
- return sendRequest(post, retType, uri, uriParams);
- }
-
- private <V> V sendJSONRequest(
- HttpRequestBase req,
- Class<V> retType,
- String uri,
- Object... uriParams) throws Exception {
- req.setHeader(HttpHeaders.ACCEPT, APPLICATION_JSON);
- req.setHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON);
- req.setHeader(HttpHeaders.CONTENT_ENCODING, "UTF-8");
- return sendRequest(req, retType, uri, uriParams);
- }
-
- private <V> V sendRequest(
- HttpRequestBase req,
- Class<V> retType,
- String uri,
- Object... uriParams) throws Exception {
- req.setURI(new URI(server.getScheme(), null, server.getHost(), server.getPort(),
- uriRoot + String.format(uri, uriParams), null, null));
- // It is no harm to set X-Requested-By when csrf protection is disabled.
- if (req instanceof HttpPost || req instanceof HttpDelete || req instanceof HttpPut
- || req instanceof HttpPatch) {
- req.addHeader("X-Requested-By", "livy");
- }
- try (CloseableHttpResponse res = client.execute(req)) {
- int status = (res.getStatusLine().getStatusCode() / 100) * 100;
- HttpEntity entity = res.getEntity();
- if (status == HttpStatus.SC_OK) {
- if (!Void.class.equals(retType)) {
- return mapper.readValue(entity.getContent(), retType);
- } else {
- return null;
- }
- } else {
- String error = EntityUtils.toString(entity);
- throw new IOException(String.format("%s: %s", res.getStatusLine().getReasonPhrase(),
- error));
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/org/apache/livy/client/http/HttpClient.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/org/apache/livy/client/http/HttpClient.java b/client-http/src/main/java/org/apache/livy/client/http/HttpClient.java
new file mode 100644
index 0000000..f40148f
--- /dev/null
+++ b/client-http/src/main/java/org/apache/livy/client/http/HttpClient.java
@@ -0,0 +1,191 @@
+/*
+ * 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.livy.client.http;
+
+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.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.*;
+
+/**
+ * What is currently missing:
+ * - monitoring of spark job IDs launched by jobs
+ */
+class HttpClient implements LivyClient {
+
+ private final HttpConf config;
+ private final LivyConnection conn;
+ private final int sessionId;
+ private final ScheduledExecutorService executor;
+ private final Serializer serializer;
+
+ private boolean stopped;
+
+ HttpClient(URI uri, HttpConf httpConf) {
+ this.config = httpConf;
+ this.stopped = false;
+
+ // If the given URI looks like it refers to an existing session, then try to connect to
+ // an existing session. Note this means that any Spark configuration in httpConf will be
+ // unused.
+ Matcher m = Pattern.compile("(.*)" + LivyConnection.SESSIONS_URI + "/([0-9]+)")
+ .matcher(uri.getPath());
+
+ try {
+ if (m.matches()) {
+ URI base = new URI(uri.getScheme(), uri.getUserInfo(), uri.getHost(), uri.getPort(),
+ m.group(1), uri.getQuery(), uri.getFragment());
+
+ this.conn = new LivyConnection(base, httpConf);
+ this.sessionId = Integer.parseInt(m.group(2));
+ conn.post(null, SessionInfo.class, "/%d/connect", sessionId);
+ } else {
+ Map<String, String> sessionConf = new HashMap<>();
+ for (Map.Entry<String, String> e : config) {
+ sessionConf.put(e.getKey(), e.getValue());
+ }
+
+ ClientMessage create = new CreateClientRequest(sessionConf);
+ this.conn = new LivyConnection(uri, httpConf);
+ this.sessionId = conn.post(create, SessionInfo.class, "/").id;
+ }
+ } catch (Exception e) {
+ throw propagate(e);
+ }
+
+ // Because we only have one connection to the server, we don't need more than a single
+ // threaded executor here.
+ this.executor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r, "HttpClient-" + sessionId);
+ t.setDaemon(true);
+ return t;
+ }
+ });
+
+ this.serializer = new Serializer();
+ }
+
+ @Override
+ public <T> JobHandle<T> submit(Job<T> job) {
+ return sendJob("submit-job", job);
+ }
+
+ @Override
+ public <T> Future<T> run(Job<T> job) {
+ return sendJob("run-job", job);
+ }
+
+ @Override
+ public synchronized void stop(boolean shutdownContext) {
+ if (!stopped) {
+ executor.shutdownNow();
+ try {
+ if (shutdownContext) {
+ conn.delete(Map.class, "/%s", sessionId);
+ }
+ } catch (Exception e) {
+ throw propagate(e);
+ } finally {
+ try {
+ conn.close();
+ } catch (Exception e) {
+ // Ignore.
+ }
+ }
+ stopped = true;
+ }
+ }
+
+ public Future<?> uploadJar(File jar) {
+ return uploadResource(jar, "upload-jar", "jar");
+ }
+
+ @Override
+ public Future<?> addJar(URI uri) {
+ return addResource("add-jar", uri);
+ }
+
+ public Future<?> uploadFile(File file) {
+ return uploadResource(file, "upload-file", "file");
+ }
+
+ @Override
+ public Future<?> addFile(URI uri) {
+ return addResource("add-file", uri);
+ }
+
+ private Future<?> uploadResource(final File file, final String command, final String paramName) {
+ Callable<Void> task = new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ conn.post(file, Void.class, paramName, "/%d/%s", sessionId, command);
+ return null;
+ }
+ };
+ return executor.submit(task);
+ }
+
+ private Future<?> addResource(final String command, final URI resource) {
+ Callable<Void> task = new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ ClientMessage msg = new AddResource(resource.toString());
+ conn.post(msg, Void.class, "/%d/%s", sessionId, command);
+ return null;
+ }
+ };
+ return executor.submit(task);
+ }
+
+ private <T> JobHandleImpl<T> sendJob(final String command, Job<T> job) {
+ final ByteBuffer serializedJob = serializer.serialize(job);
+ JobHandleImpl<T> handle = new JobHandleImpl<T>(config, conn, sessionId, executor, serializer);
+ handle.start(command, serializedJob);
+ return handle;
+ }
+
+ private RuntimeException propagate(Exception cause) {
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+
+ // For testing.
+ int getSessionId() {
+ return sessionId;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/org/apache/livy/client/http/HttpClientFactory.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/org/apache/livy/client/http/HttpClientFactory.java b/client-http/src/main/java/org/apache/livy/client/http/HttpClientFactory.java
new file mode 100644
index 0000000..622a74e
--- /dev/null
+++ b/client-http/src/main/java/org/apache/livy/client/http/HttpClientFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.livy.client.http;
+
+import java.net.URI;
+import java.util.Properties;
+
+import org.apache.livy.LivyClient;
+import org.apache.livy.LivyClientFactory;
+
+/**
+ * Factory for HTTP Livy clients.
+ */
+public final class HttpClientFactory implements LivyClientFactory {
+
+ @Override
+ public LivyClient createClient(URI uri, Properties config) {
+ if (!"http".equals(uri.getScheme()) && !"https".equals(uri.getScheme())) {
+ return null;
+ }
+
+ return new HttpClient(uri, new HttpConf(config));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/org/apache/livy/client/http/HttpConf.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/org/apache/livy/client/http/HttpConf.java b/client-http/src/main/java/org/apache/livy/client/http/HttpConf.java
new file mode 100644
index 0000000..2ae7475
--- /dev/null
+++ b/client-http/src/main/java/org/apache/livy/client/http/HttpConf.java
@@ -0,0 +1,135 @@
+/*
+ * 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.livy.client.http;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.livy.client.common.ClientConf;
+
+class HttpConf extends ClientConf<HttpConf> {
+
+ private static final String HTTP_CONF_PREFIX = "livy.client.http.";
+
+ static enum Entry implements ConfEntry {
+ CONNECTION_TIMEOUT("connection.timeout", "10s"),
+ CONNECTION_IDLE_TIMEOUT("connection.idle.timeout", "10m"),
+ SOCKET_TIMEOUT("connection.socket.timeout", "5m"),
+
+ JOB_INITIAL_POLL_INTERVAL("job.initial-poll-interval", "100ms"),
+ JOB_MAX_POLL_INTERVAL("job.max-poll-interval", "5s"),
+
+ CONTENT_COMPRESS_ENABLE("content.compress.enable", true),
+
+ // Kerberos related configuration
+ SPNEGO_ENABLED("spnego.enable", false),
+ AUTH_LOGIN_CONFIG("auth.login.config", null),
+ KRB5_DEBUG_ENABLED("krb5.debug", false),
+ KRB5_CONF("krb5.conf", null);
+
+ private final String key;
+ private final Object dflt;
+
+ private Entry(String key, Object dflt) {
+ this.key = HTTP_CONF_PREFIX + key;
+ this.dflt = dflt;
+ }
+
+ @Override
+ public String key() { return key; }
+
+ @Override
+ public Object dflt() { return dflt; }
+ }
+
+ HttpConf(Properties config) {
+ super(config);
+
+ if (getBoolean(Entry.SPNEGO_ENABLED)) {
+ if (get(Entry.AUTH_LOGIN_CONFIG ) == null) {
+ throw new IllegalArgumentException(Entry.AUTH_LOGIN_CONFIG.key + " should not be null");
+ }
+
+ if (get(Entry.KRB5_CONF) == null) {
+ throw new IllegalArgumentException(Entry.KRB5_CONF.key + " should not be null");
+ }
+
+ System.setProperty("java.security.auth.login.config", get(Entry.AUTH_LOGIN_CONFIG));
+ System.setProperty("java.security.krb5.conf", get(Entry.KRB5_CONF));
+ System.setProperty(
+ "sun.security.krb5.debug", String.valueOf(getBoolean(Entry.KRB5_DEBUG_ENABLED)));
+ // This is needed to get Kerberos credentials from the environment, instead of
+ // requiring the application to manually obtain the credentials.
+ System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
+ }
+ }
+
+ boolean isSpnegoEnabled() {
+ return getBoolean(Entry.SPNEGO_ENABLED);
+ }
+
+ private static final Map<String, DeprecatedConf> configsWithAlternatives
+ = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
+ put(HttpConf.Entry.JOB_INITIAL_POLL_INTERVAL.key, DepConf.JOB_INITIAL_POLL_INTERVAL);
+ put(HttpConf.Entry.JOB_MAX_POLL_INTERVAL.key, DepConf.JOB_MAX_POLL_INTERVAL);
+ }});
+
+ // Maps deprecated key to DeprecatedConf with the same key.
+ // There are no deprecated configs without alternatives currently.
+ private static final Map<String, DeprecatedConf> deprecatedConfigs
+ = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>());
+
+ protected Map<String, DeprecatedConf> getConfigsWithAlternatives() {
+ return configsWithAlternatives;
+ }
+
+ protected Map<String, DeprecatedConf> getDeprecatedConfigs() {
+ return deprecatedConfigs;
+ }
+
+ static enum DepConf implements DeprecatedConf {
+ JOB_INITIAL_POLL_INTERVAL("job.initial_poll_interval", "0.4"),
+ JOB_MAX_POLL_INTERVAL("job.max_poll_interval", "0.4");
+
+ private final String key;
+ private final String version;
+ private final String deprecationMessage;
+
+ private DepConf(String key, String version) {
+ this(key, version, "");
+ }
+
+ private DepConf(String key, String version, String deprecationMessage) {
+ this.key = HTTP_CONF_PREFIX + key;
+ this.version = version;
+ this.deprecationMessage = deprecationMessage;
+ }
+
+ @Override
+ public String key() { return key; }
+
+ @Override
+ public String version() { return version; }
+
+ @Override
+ public String deprecationMessage() { return deprecationMessage; }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/org/apache/livy/client/http/JobHandleImpl.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/org/apache/livy/client/http/JobHandleImpl.java b/client-http/src/main/java/org/apache/livy/client/http/JobHandleImpl.java
new file mode 100644
index 0000000..f0ffb59
--- /dev/null
+++ b/client-http/src/main/java/org/apache/livy/client/http/JobHandleImpl.java
@@ -0,0 +1,273 @@
+/*
+ * 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.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 static org.apache.livy.client.common.HttpMessages.*;
+
+class JobHandleImpl<T> extends AbstractJobHandle<T> {
+
+ private final long sessionId;
+ private final LivyConnection conn;
+ private final ScheduledExecutorService executor;
+ private final Object lock;
+ private final Serializer serializer;
+
+ private final long initialPollInterval;
+ private final long maxPollInterval;
+
+ private long jobId;
+ private T result;
+ private Throwable error;
+ private volatile boolean isDone;
+ private volatile boolean isCancelled;
+ private volatile boolean isCancelPending;
+ private volatile ScheduledFuture<?> pollTask;
+
+ JobHandleImpl(
+ HttpConf config,
+ LivyConnection conn,
+ long sessionId,
+ ScheduledExecutorService executor,
+ Serializer s) {
+ this.conn = conn;
+ this.sessionId = sessionId;
+ this.executor = executor;
+ this.lock = new Object();
+ this.serializer = s;
+ this.isDone = false;
+
+ this.initialPollInterval = config.getTimeAsMs(HttpConf.Entry.JOB_INITIAL_POLL_INTERVAL);
+ this.maxPollInterval = config.getTimeAsMs(HttpConf.Entry.JOB_MAX_POLL_INTERVAL);
+
+ if (initialPollInterval <= 0) {
+ throw new IllegalArgumentException("Invalid initial poll interval.");
+ }
+ if (maxPollInterval <= 0 || maxPollInterval < initialPollInterval) {
+ throw new IllegalArgumentException(
+ "Invalid max poll interval, or lower than initial interval.");
+ }
+
+ // The job ID is set asynchronously, and there might be a call to cancel() before it's
+ // set. So cancel() will always set the isCancelPending flag, even if there's no job
+ // ID yet. If the thread setting the job ID sees that flag, it will send a cancel request
+ // to the server. There's still a possibility that two cancel requests will be sent,
+ // but that doesn't cause any harm.
+ this.isCancelPending = false;
+ this.jobId = -1;
+ }
+
+ @Override
+ public T get() throws ExecutionException, InterruptedException {
+ try {
+ return get(true, -1, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException te) {
+ // Not gonna happen.
+ throw new RuntimeException(te);
+ }
+ }
+
+ @Override
+ public T get(long timeout, TimeUnit unit)
+ throws ExecutionException, InterruptedException, TimeoutException {
+ return get(false, timeout, unit);
+ }
+
+ @Override
+ public boolean isDone() {
+ return isDone;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return isCancelled;
+ }
+
+ @Override
+ public boolean cancel(final boolean mayInterrupt) {
+ // Do a best-effort to detect if already cancelled, but the final say is always
+ // on the server side. Don't block the caller, though.
+ if (!isCancelled && !isCancelPending) {
+ isCancelPending = true;
+ if (jobId > -1) {
+ sendCancelRequest(jobId);
+ }
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ protected T result() {
+ return result;
+ }
+
+ @Override
+ protected Throwable error() {
+ return error;
+ }
+
+ void start(final String command, final ByteBuffer serializedJob) {
+ Runnable task = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ ClientMessage msg = new SerializedJob(BufferUtils.toByteArray(serializedJob));
+ JobStatus status = conn.post(msg, JobStatus.class, "/%d/%s", sessionId, command);
+
+ if (isCancelPending) {
+ sendCancelRequest(status.id);
+ }
+
+ jobId = status.id;
+
+ pollTask = executor.schedule(new JobPollTask(initialPollInterval),
+ initialPollInterval, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ setResult(null, e, State.FAILED);
+ }
+ }
+ };
+ executor.submit(task);
+ }
+
+ private void sendCancelRequest(final long id) {
+ executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ conn.post(null, Void.class, "/%d/jobs/%d/cancel", sessionId, id);
+ } catch (Exception e) {
+ setResult(null, e, State.FAILED);
+ }
+ }
+ });
+ }
+
+ private T get(boolean waitIndefinitely, long timeout, TimeUnit unit)
+ throws ExecutionException, InterruptedException, TimeoutException {
+ if (!isDone) {
+ synchronized (lock) {
+ if (waitIndefinitely) {
+ while (!isDone) {
+ lock.wait();
+ }
+ } else {
+ long now = System.nanoTime();
+ long deadline = now + unit.toNanos(timeout);
+ while (!isDone && deadline > now) {
+ lock.wait(TimeUnit.NANOSECONDS.toMillis(deadline - now));
+ now = System.nanoTime();
+ }
+ if (!isDone) {
+ throw new TimeoutException();
+ }
+ }
+ }
+ }
+ if (isCancelled) {
+ throw new CancellationException();
+ }
+ if (error != null) {
+ throw new ExecutionException(error);
+ }
+ return result;
+ }
+
+ private void setResult(T result, Throwable error, State newState) {
+ if (!isDone) {
+ synchronized (lock) {
+ if (!isDone) {
+ this.result = result;
+ this.error = error;
+ this.isDone = true;
+ changeState(newState);
+ }
+ lock.notifyAll();
+ }
+ }
+ }
+
+ private class JobPollTask implements Runnable {
+
+ private long currentInterval;
+
+ JobPollTask(long currentInterval) {
+ this.currentInterval = currentInterval;
+ }
+
+ @Override
+ public void run() {
+ try {
+ JobStatus status = conn.get(JobStatus.class, "/%d/jobs/%d", sessionId, jobId);
+ T result = null;
+ Throwable error = null;
+ boolean finished = false;
+
+ switch (status.state) {
+ case SUCCEEDED:
+ if (status.result != null) {
+ @SuppressWarnings("unchecked")
+ T localResult = (T) serializer.deserialize(ByteBuffer.wrap(status.result));
+ result = localResult;
+ }
+ finished = true;
+ break;
+
+ case FAILED:
+ // TODO: better exception.
+ error = new RuntimeException(status.error);
+ finished = true;
+ break;
+
+ case CANCELLED:
+ isCancelled = true;
+ finished = true;
+ break;
+
+ default:
+ // Nothing to do.
+ }
+ if (finished) {
+ setResult(result, error, status.state);
+ } else if (status.state != state) {
+ changeState(status.state);
+ }
+ if (!finished) {
+ currentInterval = Math.min(currentInterval * 2, maxPollInterval);
+ pollTask = executor.schedule(this, currentInterval, TimeUnit.MILLISECONDS);
+ }
+ } catch (Exception e) {
+ setResult(null, e, State.FAILED);
+ }
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/java/org/apache/livy/client/http/LivyConnection.java
----------------------------------------------------------------------
diff --git a/client-http/src/main/java/org/apache/livy/client/http/LivyConnection.java b/client-http/src/main/java/org/apache/livy/client/http/LivyConnection.java
new file mode 100644
index 0000000..e8612bf
--- /dev/null
+++ b/client-http/src/main/java/org/apache/livy/client/http/LivyConnection.java
@@ -0,0 +1,235 @@
+/*
+ * 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.livy.client.http;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.security.Principal;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHeaders;
+import org.apache.http.HttpStatus;
+import org.apache.http.auth.AuthSchemeProvider;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.Credentials;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.AuthSchemes;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.*;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.config.Registry;
+import org.apache.http.config.RegistryBuilder;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.mime.MultipartEntityBuilder;
+import org.apache.http.entity.mime.content.FileBody;
+import org.apache.http.impl.NoConnectionReuseStrategy;
+import org.apache.http.impl.auth.SPNegoSchemeFactory;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.BasicHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+
+import static org.apache.livy.client.http.HttpConf.Entry.*;
+
+/**
+ * Abstracts a connection to the Livy server; serializes multiple requests so that we only need
+ * one active HTTP connection (to keep resource usage down).
+ */
+class LivyConnection {
+
+ static final String SESSIONS_URI = "/sessions";
+ private static final String APPLICATION_JSON = "application/json";
+
+ private final URI server;
+ private final String uriRoot;
+ private final CloseableHttpClient client;
+ private final ObjectMapper mapper;
+
+ LivyConnection(URI uri, final HttpConf config) {
+ HttpClientContext ctx = HttpClientContext.create();
+ int port = uri.getPort() > 0 ? uri.getPort() : 8998;
+
+ String path = uri.getPath() != null ? uri.getPath() : "";
+ this.uriRoot = path + SESSIONS_URI;
+
+ RequestConfig reqConfig = new RequestConfig() {
+ @Override
+ public int getConnectTimeout() {
+ return (int) config.getTimeAsMs(CONNECTION_TIMEOUT);
+ }
+
+ @Override
+ public int getSocketTimeout() {
+ return (int) config.getTimeAsMs(SOCKET_TIMEOUT);
+ }
+
+ @Override
+ public boolean isAuthenticationEnabled() {
+ return true;
+ }
+
+ @Override
+ public boolean isContentCompressionEnabled() {
+ return config.getBoolean(CONTENT_COMPRESS_ENABLE);
+ }
+ };
+
+ Credentials credentials;
+ // If user info is specified in the url, pass them to the CredentialsProvider.
+ if (uri.getUserInfo() != null) {
+ String[] userInfo = uri.getUserInfo().split(":");
+ if (userInfo.length < 1) {
+ throw new IllegalArgumentException("Malformed user info in the url.");
+ }
+ try {
+ String username = URLDecoder.decode(userInfo[0], StandardCharsets.UTF_8.name());
+ String password = "";
+ if (userInfo.length > 1) {
+ password = URLDecoder.decode(userInfo[1], StandardCharsets.UTF_8.name());
+ }
+ credentials = new UsernamePasswordCredentials(username, password);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("User info in the url contains bad characters.", e);
+ }
+ } else {
+ credentials = new Credentials() {
+ @Override
+ public String getPassword() {
+ return null;
+ }
+
+ @Override
+ public Principal getUserPrincipal() {
+ return null;
+ }
+ };
+ }
+
+ CredentialsProvider credsProvider = new BasicCredentialsProvider();
+ credsProvider.setCredentials(AuthScope.ANY, credentials);
+
+ HttpClientBuilder builder = HttpClientBuilder.create()
+ .disableAutomaticRetries()
+ .evictExpiredConnections()
+ .evictIdleConnections(config.getTimeAsMs(CONNECTION_IDLE_TIMEOUT), TimeUnit.MILLISECONDS)
+ .setConnectionManager(new BasicHttpClientConnectionManager())
+ .setConnectionReuseStrategy(new NoConnectionReuseStrategy())
+ .setDefaultRequestConfig(reqConfig)
+ .setMaxConnTotal(1)
+ .setDefaultCredentialsProvider(credsProvider)
+ .setUserAgent("livy-client-http");
+
+ if (config.isSpnegoEnabled()) {
+ Registry<AuthSchemeProvider> authSchemeProviderRegistry =
+ RegistryBuilder.<AuthSchemeProvider>create()
+ .register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory())
+ .build();
+ builder.setDefaultAuthSchemeRegistry(authSchemeProviderRegistry);
+ }
+
+ this.server = uri;
+ this.client = builder.build();
+ this.mapper = new ObjectMapper();
+ }
+
+ synchronized void close() throws IOException {
+ client.close();
+ }
+
+ synchronized <V> V delete(Class<V> retType, String uri, Object... uriParams) throws Exception {
+ return sendJSONRequest(new HttpDelete(), retType, uri, uriParams);
+ }
+
+ synchronized <V> V get(Class<V> retType, String uri, Object... uriParams) throws Exception {
+ return sendJSONRequest(new HttpGet(), retType, uri, uriParams);
+ }
+
+ synchronized <V> V post(
+ Object body,
+ Class<V> retType,
+ String uri,
+ Object... uriParams) throws Exception {
+ HttpPost post = new HttpPost();
+ if (body != null) {
+ byte[] bodyBytes = mapper.writeValueAsBytes(body);
+ post.setEntity(new ByteArrayEntity(bodyBytes));
+ }
+ return sendJSONRequest(post, retType, uri, uriParams);
+ }
+
+ synchronized <V> V post(
+ File f,
+ Class<V> retType,
+ String paramName,
+ String uri,
+ Object... uriParams) throws Exception {
+ HttpPost post = new HttpPost();
+ MultipartEntityBuilder builder = MultipartEntityBuilder.create();
+ builder.addPart(paramName, new FileBody(f));
+ post.setEntity(builder.build());
+ return sendRequest(post, retType, uri, uriParams);
+ }
+
+ private <V> V sendJSONRequest(
+ HttpRequestBase req,
+ Class<V> retType,
+ String uri,
+ Object... uriParams) throws Exception {
+ req.setHeader(HttpHeaders.ACCEPT, APPLICATION_JSON);
+ req.setHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON);
+ req.setHeader(HttpHeaders.CONTENT_ENCODING, "UTF-8");
+ return sendRequest(req, retType, uri, uriParams);
+ }
+
+ private <V> V sendRequest(
+ HttpRequestBase req,
+ Class<V> retType,
+ String uri,
+ Object... uriParams) throws Exception {
+ req.setURI(new URI(server.getScheme(), null, server.getHost(), server.getPort(),
+ uriRoot + String.format(uri, uriParams), null, null));
+ // It is no harm to set X-Requested-By when csrf protection is disabled.
+ if (req instanceof HttpPost || req instanceof HttpDelete || req instanceof HttpPut
+ || req instanceof HttpPatch) {
+ req.addHeader("X-Requested-By", "livy");
+ }
+ try (CloseableHttpResponse res = client.execute(req)) {
+ int status = (res.getStatusLine().getStatusCode() / 100) * 100;
+ HttpEntity entity = res.getEntity();
+ if (status == HttpStatus.SC_OK) {
+ if (!Void.class.equals(retType)) {
+ return mapper.readValue(entity.getContent(), retType);
+ } else {
+ return null;
+ }
+ } else {
+ String error = EntityUtils.toString(entity);
+ throw new IOException(String.format("%s: %s", res.getStatusLine().getReasonPhrase(),
+ error));
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/resources/META-INF/services/com.cloudera.livy.LivyClientFactory
----------------------------------------------------------------------
diff --git a/client-http/src/main/resources/META-INF/services/com.cloudera.livy.LivyClientFactory b/client-http/src/main/resources/META-INF/services/com.cloudera.livy.LivyClientFactory
deleted file mode 100644
index b186c05..0000000
--- a/client-http/src/main/resources/META-INF/services/com.cloudera.livy.LivyClientFactory
+++ /dev/null
@@ -1 +0,0 @@
-com.cloudera.livy.client.http.HttpClientFactory
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/main/resources/META-INF/services/org.apache.livy.LivyClientFactory
----------------------------------------------------------------------
diff --git a/client-http/src/main/resources/META-INF/services/org.apache.livy.LivyClientFactory b/client-http/src/main/resources/META-INF/services/org.apache.livy.LivyClientFactory
new file mode 100644
index 0000000..5dd53aa
--- /dev/null
+++ b/client-http/src/main/resources/META-INF/services/org.apache.livy.LivyClientFactory
@@ -0,0 +1 @@
+org.apache.livy.client.http.HttpClientFactory
http://git-wip-us.apache.org/repos/asf/incubator-livy/blob/412ccc8f/client-http/src/test/scala/com/cloudera/livy/client/http/HttpClientSpec.scala
----------------------------------------------------------------------
diff --git a/client-http/src/test/scala/com/cloudera/livy/client/http/HttpClientSpec.scala b/client-http/src/test/scala/com/cloudera/livy/client/http/HttpClientSpec.scala
deleted file mode 100644
index f3ebcd5..0000000
--- a/client-http/src/test/scala/com/cloudera/livy/client/http/HttpClientSpec.scala
+++ /dev/null
@@ -1,290 +0,0 @@
-/*
- * 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 com.cloudera.livy.client.http
-
-import java.io.{File, InputStream}
-import java.net.{InetAddress, URI}
-import java.nio.file.{Files, Paths}
-import java.util.concurrent.{Future => JFuture, _}
-import java.util.concurrent.atomic.AtomicLong
-import javax.servlet.ServletContext
-import javax.servlet.http.HttpServletRequest
-
-import scala.concurrent.{ExecutionContext, Future}
-
-import org.mockito.ArgumentCaptor
-import org.mockito.Matchers.{eq => meq, _}
-import org.mockito.Mockito._
-import org.scalatest.{BeforeAndAfterAll, FunSpecLike}
-import org.scalatra.LifeCycle
-import org.scalatra.servlet.ScalatraListener
-
-import com.cloudera.livy._
-import com.cloudera.livy.client.common.{BufferUtils, Serializer}
-import com.cloudera.livy.client.common.HttpMessages._
-import com.cloudera.livy.server.WebServer
-import com.cloudera.livy.server.interactive.{InteractiveSession, InteractiveSessionServlet}
-import com.cloudera.livy.server.recovery.SessionStore
-import com.cloudera.livy.sessions.{InteractiveSessionManager, SessionState, Spark}
-import com.cloudera.livy.test.jobs.Echo
-import com.cloudera.livy.utils.AppInfo
-
-/**
- * The test for the HTTP client is written in Scala so we can reuse the code in the livy-server
- * module, which implements the client session backend. The client servlet has some functionality
- * overridden to avoid creating sub-processes for each seession.
- */
-class HttpClientSpec extends FunSpecLike with BeforeAndAfterAll with LivyBaseUnitTestSuite {
-
- import HttpClientSpec._
-
- private val TIMEOUT_S = 10
- private val ID_GENERATOR = new AtomicLong()
- private val serializer = new Serializer()
-
- private var server: WebServer = _
- private var client: LivyClient = _
-
- override def beforeAll(): Unit = {
- super.beforeAll()
- server = new WebServer(new LivyConf(), "0.0.0.0", 0)
-
- server.context.setResourceBase("src/main/com/cloudera/livy/server")
- server.context.setInitParameter(ScalatraListener.LifeCycleKey,
- classOf[HttpClientTestBootstrap].getCanonicalName)
- server.context.addEventListener(new ScalatraListener)
-
- server.start()
- }
-
- override def afterAll(): Unit = {
- super.afterAll()
- if (server != null) {
- server.stop()
- server = null
- }
- if (client != null) {
- client.stop(true)
- client = null
- }
- session = null
- }
-
- describe("HTTP client library") {
-
- it("should create clients") {
- // WebServer does this internally instead of respecting "0.0.0.0", so try to use the same
- // address.
- val uri = s"http://${InetAddress.getLocalHost.getHostAddress}:${server.port}/"
- client = new LivyClientBuilder(false).setURI(new URI(uri)).build()
- }
-
- withClient("should run and monitor asynchronous jobs") {
- testJob(false)
- }
-
- withClient("should propagate errors from jobs") {
- val errorMessage = "This job throws an error."
- val (jobId, handle) = runJob(false, { id => Seq(
- new JobStatus(id, JobHandle.State.FAILED, null, errorMessage))
- })
-
- val error = intercept[ExecutionException] {
- handle.get(TIMEOUT_S, TimeUnit.SECONDS)
- }
- assert(error.getCause() != null)
- assert(error.getCause().getMessage().indexOf(errorMessage) >= 0)
- verify(session, times(1)).jobStatus(meq(jobId))
- }
-
- withClient("should run and monitor synchronous jobs") {
- testJob(false)
- }
-
- withClient("should add files and jars") {
- val furi = new URI("hdfs:file")
- val juri = new URI("hdfs:jar")
-
- client.addFile(furi).get(TIMEOUT_S, TimeUnit.SECONDS)
- client.addJar(juri).get(TIMEOUT_S, TimeUnit.SECONDS)
-
- verify(session, times(1)).addFile(meq(furi))
- verify(session, times(1)).addJar(meq(juri))
- }
-
- withClient("should upload files and jars") {
- uploadAndVerify("file")
- uploadAndVerify("jar")
- }
-
- withClient("should cancel jobs") {
- val (jobId, handle) = runJob(false, { id => Seq(
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.CANCELLED, null, null))
- })
- handle.cancel(true)
-
- intercept[CancellationException] {
- handle.get(TIMEOUT_S, TimeUnit.SECONDS)
- }
-
- verify(session, times(1)).cancelJob(meq(jobId))
- }
-
- withClient("should notify listeners of job completion") {
- val (jobId, handle) = runJob(false, { id => Seq(
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.SUCCEEDED, serialize(id), null))
- })
-
- val listener = mock(classOf[JobHandle.Listener[Long]])
- handle.asInstanceOf[JobHandle[Long]].addListener(listener)
-
- assert(handle.get(TIMEOUT_S, TimeUnit.SECONDS) === jobId)
- verify(listener, times(1)).onJobSucceeded(any(), any())
- }
-
- withClient("should time out handle get() call") {
- // JobHandleImpl does exponential backoff checking the result of a job. Given an initial
- // wait of 100ms, 4 iterations should result in a wait of 800ms, so the handle should at that
- // point timeout a wait of 100ms.
- val (jobId, handle) = runJob(false, { id => Seq(
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.SUCCEEDED, serialize(id), null))
- })
-
- intercept[TimeoutException] {
- handle.get(100, TimeUnit.MILLISECONDS)
- }
-
- assert(handle.get(TIMEOUT_S, TimeUnit.SECONDS) === jobId)
- }
-
- withClient("should handle null responses") {
- testJob(false, response = Some(null))
- }
-
- withClient("should connect to existing sessions") {
- var sid = client.asInstanceOf[HttpClient].getSessionId()
- val uri = s"http://${InetAddress.getLocalHost.getHostAddress}:${server.port}" +
- s"${LivyConnection.SESSIONS_URI}/$sid"
- val newClient = new LivyClientBuilder(false).setURI(new URI(uri)).build()
- newClient.stop(false)
- verify(session, never()).stop()
- }
-
- withClient("should tear down clients") {
- client.stop(true)
- verify(session, times(1)).stop()
- }
-
- }
-
- private def uploadAndVerify(cmd: String): Unit = {
- val f = File.createTempFile("uploadTestFile", cmd)
- val expectedStr = "Test data"
- val expectedData = expectedStr.getBytes()
- Files.write(Paths.get(f.getAbsolutePath), expectedData)
- val b = new Array[Byte](expectedData.length)
- val captor = ArgumentCaptor.forClass(classOf[InputStream])
- if (cmd == "file") {
- client.uploadFile(f).get(TIMEOUT_S, TimeUnit.SECONDS)
- verify(session, times(1)).addFile(captor.capture(), meq(f.getName))
- } else {
- client.uploadJar(f).get(TIMEOUT_S, TimeUnit.SECONDS)
- verify(session, times(1)).addJar(captor.capture(), meq(f.getName))
- }
- captor.getValue.read(b)
- assert(expectedStr === new String(b))
- }
-
- private def runJob(sync: Boolean, genStatusFn: Long => Seq[JobStatus]): (Long, JFuture[Int]) = {
- val jobId = java.lang.Long.valueOf(ID_GENERATOR.incrementAndGet())
- when(session.submitJob(any(classOf[Array[Byte]]))).thenReturn(jobId)
-
- val statuses = genStatusFn(jobId)
- val first = statuses.head
- val remaining = statuses.drop(1)
- when(session.jobStatus(meq(jobId))).thenReturn(first, remaining: _*)
-
- val job = new Echo(42)
- val handle = if (sync) client.run(job) else client.submit(job)
- (jobId, handle)
- }
-
- private def testJob(sync: Boolean, response: Option[Any] = None): Unit = {
- val (jobId, handle) = runJob(sync, { id => Seq(
- new JobStatus(id, JobHandle.State.STARTED, null, null),
- new JobStatus(id, JobHandle.State.SUCCEEDED, serialize(response.getOrElse(id)), null))
- })
- assert(handle.get(TIMEOUT_S, TimeUnit.SECONDS) === response.getOrElse(jobId))
- verify(session, times(2)).jobStatus(meq(jobId))
- }
-
- private def withClient(desc: String)(fn: => Unit): Unit = {
- it(desc) {
- assume(client != null, "No active client.")
- fn
- }
- }
-
- def serialize(value: Any): Array[Byte] = {
- BufferUtils.toByteArray(serializer.serialize(value))
- }
-
-}
-
-private object HttpClientSpec {
-
- // Hack warning: keep the session object available so that individual tests can mock
- // the desired behavior before making requests to the server.
- var session: InteractiveSession = _
-
-}
-
-private class HttpClientTestBootstrap extends LifeCycle {
-
- private implicit def executor: ExecutionContext = ExecutionContext.global
-
- override def init(context: ServletContext): Unit = {
- val conf = new LivyConf()
- val stateStore = mock(classOf[SessionStore])
- val sessionManager = new InteractiveSessionManager(conf, stateStore, Some(Seq.empty))
- val servlet = new InteractiveSessionServlet(sessionManager, stateStore, conf) {
- override protected def createSession(req: HttpServletRequest): InteractiveSession = {
- val session = mock(classOf[InteractiveSession])
- val id = sessionManager.nextId()
- when(session.id).thenReturn(id)
- when(session.appId).thenReturn(None)
- when(session.appInfo).thenReturn(AppInfo())
- when(session.state).thenReturn(SessionState.Idle())
- when(session.proxyUser).thenReturn(None)
- when(session.kind).thenReturn(Spark())
- when(session.stop()).thenReturn(Future.successful(()))
- require(HttpClientSpec.session == null, "Session already created?")
- HttpClientSpec.session = session
- session
- }
- }
-
- context.mount(servlet, s"${LivyConnection.SESSIONS_URI}/*")
- }
-
-}