Caused by: java.io.IOException: Unable to connect to provided ports 10000~10010

.gitignore

@@ -11,7 +11,11 @@
 .project
 .classpath
 .scala_dependencies
+conf/*.conf
+conf/*.properties
+conf/*.sh
 lib_managed/
+logs/
 src_managed/
 target/
 reports/

.travis.yml

@@ -3,8 +3,12 @@ dist: trusty
 language: scala
 
 env:
-  - MVN_FLAG=-Pspark-1.6
-  - MVN_FLAG=-Pspark-2.0
+  - MVN_FLAG='-Pspark-1.6 -DskipTests'
+  - MVN_FLAG='-Pspark-2.0 -DskipTests'
+  - MVN_FLAG='-Pspark-2.1 -DskipTests'
+  - MVN_FLAG='-Pspark-1.6 -DskipITs'
+  - MVN_FLAG='-Pspark-2.0 -DskipITs'
+  - MVN_FLAG='-Pspark-2.1 -DskipITs'
 
 jdk:
   - oraclejdk7
@@ -24,18 +28,19 @@ cache:
 
 before_install:
   - sudo apt-get -y install python3-pip python-dev
+  - sudo apt-get -y install libkrb5-dev
   - sudo apt-get -y remove python-setuptools
-  - pip install --user --upgrade pip setuptools
-  - pip3 install --user --upgrade pip setuptools
+  - pip install --user --upgrade pip "setuptools < 36"
+  - pip3 install --user --upgrade pip "setuptools < 36"
   - pip install --user codecov cloudpickle
   - pip3 install --user cloudpickle
 
 install:
   - mvn $MVN_FLAG install -Dskip -DskipTests -DskipITs -Dmaven.javadoc.skip=true -B -V
 
 before_script:
-  - pip install --user requests pytest flaky flake8
-  - pip3 install --user requests pytest flaky
+  - pip install --user requests pytest flaky flake8 requests-kerberos
+  - pip3 install --user requests pytest flaky requests-kerberos
 
 script:
   - mvn $MVN_FLAG verify -e

api/pom.xml

@@ -20,12 +20,12 @@
   <parent>
     <groupId>com.cloudera.livy</groupId>
     <artifactId>livy-main</artifactId>
-    <version>0.3.0-SNAPSHOT</version>
+    <version>0.4.0-SNAPSHOT</version>
   </parent>
 
   <groupId>com.cloudera.livy</groupId>
   <artifactId>livy-api</artifactId>
-  <version>0.3.0-SNAPSHOT</version>
+  <version>0.4.0-SNAPSHOT</version>
   <packaging>jar</packaging>
 
   <properties>

api/src/main/java/com/cloudera/livy/JobContext.java

@@ -38,10 +38,10 @@ public interface JobContext {
   /** The shared SparkContext instance. */
   JavaSparkContext sc();
 
-  /** The shared SQLContext inststance. */
+  /** The shared SQLContext instance. */
   SQLContext sqlctx();
 
-  /** The shared HiveContext inststance. */
+  /** The shared HiveContext instance. */
   HiveContext hivectx();
 
   /** Returns the JavaStreamingContext which has already been created. */
@@ -63,4 +63,8 @@ public interface JobContext {
    */
   File getLocalTmpDir();
 
+  /**
+   * Returns SparkSession if it existed, otherwise throws Exception.
+   */
+  <E> E sparkSession() throws Exception;
 }

assembly/pom.xml

@@ -21,12 +21,12 @@
   <parent>
     <groupId>com.cloudera.livy</groupId>
     <artifactId>livy-main</artifactId>
-    <version>0.3.0-SNAPSHOT</version>
+    <version>0.4.0-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
   <artifactId>livy-assembly</artifactId>
-  <version>0.3.0-SNAPSHOT</version>
+  <version>0.4.0-SNAPSHOT</version>
   <packaging>pom</packaging>
 
   <properties>

bin/livy-server

@@ -108,7 +108,7 @@ start_livy_server() {
     if [ -f "$pid" ]; then
       TARGET_ID="$(cat "$pid")"
       if [[ $(ps -p "$TARGET_ID" -o comm=) =~ "java" ]]; then
-        echo "livy_server running as process $TARGET_ID.  Stop it first."
+        echo "livy-server running as process $TARGET_ID.  Stop it first."
         exit 1
       fi
     fi
@@ -125,6 +125,7 @@ start_livy_server() {
       tail -2 "$log" | sed 's/^/  /'
       echo "full log in $log"
       rm -rf "$pid"
+      exit 1
     fi
   fi
 }
@@ -146,13 +147,26 @@ case $option in
     if [ -f $pid ]; then
       TARGET_ID="$(cat "$pid")"
       if [[ $(ps -p "$TARGET_ID" -o comm=) =~ "java" ]]; then
-        echo "stopping livy_server"
+        echo "stopping livy-server"
         kill "$TARGET_ID" && rm -f "$pid"
       else
-        echo "no livy_server to stop"
+        echo "no livy-server to stop"
       fi
     else
-      echo "no livy_server to stop"
+      echo "no livy-server to stop"
+    fi
+    ;;
+
+  (status)
+    if [ -f "$pid" ]; then
+      TARGET_ID="$(cat "$pid")"
+      if [[ $(ps -p "$TARGET_ID" -o comm=) =~ "java" ]]; then
+        echo "livy-server is running (pid: $TARGET_ID)"
+      else
+        echo "livy-server is not running"
+      fi
+    else
+      echo "livy-server is not running"
     fi
     ;;
 

client-common/pom.xml

@@ -21,12 +21,12 @@
   <parent>
     <groupId>com.cloudera.livy</groupId>
     <artifactId>livy-main</artifactId>
-    <version>0.3.0-SNAPSHOT</version>
+    <version>0.4.0-SNAPSHOT</version>
   </parent>
 
   <groupId>com.cloudera.livy</groupId>
   <artifactId>livy-client-common</artifactId>
-  <version>0.3.0-SNAPSHOT</version>
+  <version>0.4.0-SNAPSHOT</version>
   <packaging>jar</packaging>
 
   <dependencies>
@@ -45,5 +45,10 @@
       <artifactId>jackson-databind</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <scope>provided</scope>
+    </dependency>
   </dependencies>
 </project>

client-common/src/main/java/com/cloudera/livy/client/common/ClientConf.java

@@ -28,6 +28,9 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.cloudera.livy.annotations.Private;
 
 /**
@@ -37,6 +40,8 @@
 public abstract class ClientConf<T extends ClientConf>
   implements Iterable<Map.Entry<String, String>> {
 
+  protected Logger LOG = LoggerFactory.getLogger(getClass());
+
   public static interface ConfEntry {
 
     /** The key in the configuration file. */
@@ -71,24 +76,37 @@ protected ClientConf(Properties config) {
     this.config = new ConcurrentHashMap<>();
     if (config != null) {
       for (String key : config.stringPropertyNames()) {
+        logDeprecationWarning(key);
         this.config.put(key, config.getProperty(key));
       }
     }
   }
 
   public String get(String key) {
-    return config.get(key);
+    String val = config.get(key);
+    if (val != null) {
+      return val;
+    }
+    DeprecatedConf depConf = getConfigsWithAlternatives().get(key);
+    if (depConf != null) {
+      return config.get(depConf.key());
+    } else {
+      return val;
+    }
   }
 
   @SuppressWarnings("unchecked")
   public T set(String key, String value) {
+    logDeprecationWarning(key);
     config.put(key, value);
     return (T) this;
   }
 
   @SuppressWarnings("unchecked")
   public T setIfMissing(String key, String value) {
-    config.putIfAbsent(key, value);
+    if (config.putIfAbsent(key, value) == null) {
+      logDeprecationWarning(key);
+    }
     return (T) this;
   }
 
@@ -163,6 +181,7 @@ public T set(ConfEntry e, Object value) {
     if (value == null) {
       config.remove(e.key());
     } else {
+      logDeprecationWarning(e.key());
       config.put(e.key(), value.toString());
     }
     return (T) this;
@@ -176,7 +195,7 @@ public Iterator<Map.Entry<String, String>> iterator() {
   private String get(ConfEntry e, Class<?> requestedType) {
     check(getType(e.dflt()).equals(requestedType), "Invalid type conversion requested for %s.",
       e.key());
-    return config.get(e.key());
+    return this.get(e.key());
   }
 
   private boolean typesMatch(Object test, Object expected) {
@@ -193,4 +212,40 @@ private void check(boolean test, String message, Object... args) {
     }
   }
 
+  /** Logs a warning message if the given config key is deprecated. */
+  private void logDeprecationWarning(String key) {
+    DeprecatedConf altConfs = getConfigsWithAlternatives().get(key);
+    if (altConfs != null) {
+      LOG.warn("The configuration key " + altConfs.key() + " has been deprecated as of Livy "
+        + altConfs.version() + " and may be removed in the future. Please use the new key "
+        + key + " instead.");
+      return;
+    }
+
+    DeprecatedConf depConfs = getDeprecatedConfigs().get(key);
+    if (depConfs != null) {
+      LOG.warn("The configuration key " + depConfs.key() + " has been deprecated as of Livy "
+        + depConfs.version() + " and may be removed in the future. "
+        + depConfs.deprecationMessage());
+    }
+  }
+
+  /** Maps valid key to DeprecatedConf with the deprecated key. */
+  protected abstract Map<String, DeprecatedConf> getConfigsWithAlternatives();
+
+  /** Maps deprecated key to DeprecatedConf with the same key. */
+  protected abstract Map<String, DeprecatedConf> getDeprecatedConfigs();
+
+  public static interface DeprecatedConf {
+
+    /** The key in the configuration file. */
+    String key();
+
+    /** The Livy version in which the key was deprecated. */
+    String version();
+
+    /** Message to include in the deprecation warning for configs without alternatives */
+    String deprecationMessage();
+  }
+
 }

client-common/src/test/java/com/cloudera/livy/client/common/TestClientConf.java

@@ -18,6 +18,8 @@
 
 package com.cloudera.livy.client.common;
 
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 
@@ -122,6 +124,31 @@ public void testTimeWithoutDefault() {
     conf.getTimeAsMs(TestConf.Entry.TIME_NO_DEFAULT);
   }
 
+
+  @Test
+  public void testDeprecation() {
+    TestConf conf = new TestConf(null);
+
+    assertNull(conf.get("depKey"));
+    assertNull(conf.get("dep_alt"));
+    assertNull(conf.get("new-key"));
+    assertEquals("value", conf.get(TestConf.Entry.NEW_CONF));
+
+    TestConf depProps = new TestConf(null);
+    depProps.set("depKey", "dep-val");
+    depProps.set("dep_alt", "alt-val");
+    conf.setAll(depProps);
+    assertEquals("dep-val", conf.get("depKey"));
+    assertEquals("alt-val", conf.get("dep_alt"));
+    assertEquals("alt-val", conf.get(TestConf.Entry.NEW_CONF));
+    assertEquals("alt-val", conf.get("new-key"));
+
+    conf.set("new-key", "new-val");
+    assertEquals("new-val", conf.get(TestConf.Entry.NEW_CONF));
+    assertEquals("alt-val", conf.get("dep_alt"));
+    assertEquals("new-val", conf.get("new-key"));
+  }
+
   private static class TestConf extends ClientConf<TestConf> {
 
     static enum Entry implements ConfEntry {
@@ -131,7 +158,8 @@ static enum Entry implements ConfEntry {
       INT("int", 42),
       LONG("long", 84L),
       TIME("time", "168ms"),
-      TIME_NO_DEFAULT("time2", null);
+      TIME_NO_DEFAULT("time2", null),
+      NEW_CONF("new-key", "value");
 
       private final String key;
       private final Object dflt;
@@ -153,6 +181,51 @@ private Entry(String key, Object dflt) {
       super(p);
     }
 
+    private static final Map<String, DeprecatedConf> configsWithAlternatives
+      = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
+      put(TestConf.Entry.NEW_CONF.key, DepConf.DEP_WITH_ALT);
+    }});
+
+    private static final Map<String, DeprecatedConf> deprecatedConfigs
+      = Collections.unmodifiableMap(new HashMap<String, DeprecatedConf>() {{
+      put(DepConf.DEP_NO_ALT.key, DepConf.DEP_NO_ALT);
+    }});
+
+    protected Map<String, DeprecatedConf> getConfigsWithAlternatives() {
+      return configsWithAlternatives;
+    }
+
+    protected Map<String, DeprecatedConf> getDeprecatedConfigs() {
+      return deprecatedConfigs;
+    }
+
+    static enum DepConf implements DeprecatedConf {
+      DEP_WITH_ALT("dep_alt", "0.4"),
+      DEP_NO_ALT("depKey", "1.0");
+
+      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 = 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; }
+    }
   }
 
 }