Support Samza on Kubernetes. Contributed by Weiqing Yang &Jian He

Author: jian.h

build.gradle

@@ -564,6 +564,27 @@ project(":samza-yarn_$scalaSuffix") {
   jar.dependsOn("lesscss")
 }
 
+project(":samza-kubernetes_$scalaSuffix") {
+  apply plugin: 'java'
+
+  dependencies {
+    compile project(':samza-api')
+    compile project(":samza-core_$scalaSuffix")
+    compile "org.codehaus.jackson:jackson-core-asl:1.9.7"
+    compile group: 'io.fabric8', name: 'kubernetes-client', version: kubernetesJavaClientVersion
+    testCompile "junit:junit:$junitVersion"
+    testCompile "org.mockito:mockito-core:$mockitoVersion"
+  }
+
+  tasks.create(name: "releaseKubeTarGz", dependsOn: configurations.archives.artifacts, type: Tar) {
+    into "samza-kubernetes-${version}"
+    compression = Compression.GZIP
+    from(configurations.runtime) { into("lib/") }
+    from(configurations.archives.artifacts.files) { into("lib/") }
+    duplicatesStrategy 'exclude'
+  }
+}
+
 project(":samza-shell") {
   apply plugin: 'java'
 

gradle/dependency-versions.gradle

@@ -49,4 +49,5 @@
   failsafeVersion = "1.1.0"
   jlineVersion = "3.8.2"
   jnaVersion = "4.5.1"
+  kubernetesJavaClientVersion = "4.1.3"
 }

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,5 @@
-#Mon Oct 31 23:13:44 PDT 2016
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.8.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-2.8-bin.zip

gradle/wrapper/gradle-wrapper.properties

@@ -76,7 +76,7 @@ public class ContainerProcessManager implements ClusterResourceManager.Callback
    * The Allocator matches requests to resources and executes processes.
    */
   private final AbstractContainerAllocator containerAllocator;
-  private final Thread allocatorThread;
+  private Thread allocatorThread = null;
 
   // The StandbyContainerManager manages standby-aware allocation and failover of containers
   private final Optional<StandbyContainerManager> standbyContainerManager;
@@ -146,8 +146,9 @@ public ContainerProcessManager(Config config,
     } else {
       this.containerAllocator = new ContainerAllocator(clusterResourceManager, config, state);
     }
-
-    this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
+    if (shouldStartAllocateThread()) {
+      this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
+    }
     log.info("finished initialization of samza task manager");
 
   }
@@ -174,10 +175,18 @@ public ContainerProcessManager(Config config,
       this.containerAllocator = new ContainerAllocator(clusterResourceManager, config, state);
     }
 
-    this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
+    if (shouldStartAllocateThread()) {
+      this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
+    }
     log.info("finished initialization of samza task manager");
   }
 
+  // In Kubernetes, the pod will be started by kubelet automatically once it is allocated, it does not need a
+  // separate thread to keep polling the allocated resources to start the container.
+  public boolean shouldStartAllocateThread() {
+    return !clusterResourceManager.getClass().getSimpleName().equals("KubeClusterResourceManager");
+  }
+
   public boolean shouldShutdown() {
     log.debug(" TaskManager state: Completed containers: {}, Configured containers: {}, Is there too many FailedContainers: {}, Is AllocatorThread alive: {} ",
       state.completedContainers.get(), state.containerCount, tooManyFailedContainers ? "yes" : "no", allocatorThread.isAlive() ? "yes" : "no");
@@ -206,20 +215,24 @@ public void start() {
 
     // Start container allocator thread
     log.info("Starting the container allocator thread");
-    allocatorThread.start();
+    if (allocatorThread != null) {
+      allocatorThread.start();
+    }
   }
 
   public void stop() {
     log.info("Invoked stop of the Samza container process manager");
 
     // Shutdown allocator thread
     containerAllocator.stop();
-    try {
-      allocatorThread.join();
-      log.info("Stopped container allocator");
-    } catch (InterruptedException ie) {
-      log.error("Allocator Thread join() threw an interrupted exception", ie);
-      Thread.currentThread().interrupt();
+    if (allocatorThread != null) {
+      try {
+        allocatorThread.join();
+        log.info("Stopped container allocator");
+      } catch (InterruptedException ie) {
+        log.error("Allocator Thread join() threw an interrupted exception", ie);
+        Thread.currentThread().interrupt();
+      }
     }
 
     if (metrics != null) {

gradlew

@@ -23,11 +23,6 @@ import java.util
 import java.util.concurrent.atomic.AtomicReference
 
 import org.apache.samza.Partition
-import org.apache.samza.config.JobConfig.Config2Job
-import org.apache.samza.config.SystemConfig.Config2System
-import org.apache.samza.config.TaskConfig.Config2Task
-import org.apache.samza.config.{Config, _}
-import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory
 import org.apache.samza.config._
 import org.apache.samza.config.JobConfig.Config2Job
 import org.apache.samza.config.SystemConfig.Config2System

gradlew.bat

@@ -136,4 +136,14 @@ class HttpServer(
       throw new SamzaException("HttpServer is not currently running, so URLs are not available for it.")
     }
   }
+
+  def getIpUrl = {
+    if (running) {
+      val runningPort = server.getConnectors()(0).asInstanceOf[NetworkConnector].getLocalPort()
+
+      new URL("http://" + Util.getLocalHost.getHostAddress + ":" + runningPort + rootPath)
+    } else {
+      throw new SamzaException("HttpServer is not currently running, so URLs are not available for it.")
+    }
+  }
 }

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java

@@ -55,6 +55,7 @@ object HttpUtil {
       (exception, loop) => {
         exception match {
           case ioe: IOException => {
+            error(ioe)
             warn("Error getting response from Job coordinator server. received IOException: %s. Retrying..." format ioe.getClass)
             httpConn = getHttpConnection(url, timeout)
           }