You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2016/04/28 00:47:53 UTC
samza git commit: SAMZA-940: fix
TestProcessJob.testProcessJobKillShouldWork fails occasionally
Repository: samza
Updated Branches:
refs/heads/master 84cf4ae0a -> efa50a790
SAMZA-940: fix TestProcessJob.testProcessJobKillShouldWork fails occasionally
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/efa50a79
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/efa50a79
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/efa50a79
Branch: refs/heads/master
Commit: efa50a7905a0027de15d1c7fe9fea6b256341690
Parents: 84cf4ae
Author: Jacob Maes <ja...@gmail.com>
Authored: Wed Apr 27 15:45:36 2016 -0700
Committer: Yi Pan (Data Infrastructure) <ni...@gmail.com>
Committed: Wed Apr 27 15:45:36 2016 -0700
----------------------------------------------------------------------
.../apache/samza/job/util/ProcessKiller.java | 48 ++++++++++++++++++++
.../org/apache/samza/job/local/ProcessJob.scala | 3 +-
2 files changed, 50 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/efa50a79/samza-core/src/main/java/org/apache/samza/job/util/ProcessKiller.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/job/util/ProcessKiller.java b/samza-core/src/main/java/org/apache/samza/job/util/ProcessKiller.java
new file mode 100644
index 0000000..62a0703
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/job/util/ProcessKiller.java
@@ -0,0 +1,48 @@
+/*
+ * 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.samza.job.util;
+
+import java.lang.reflect.Method;
+
+
+/**
+ * Kills a {@link Process} consistently, independent of Java version.
+ */
+public class ProcessKiller {
+ /**
+ * Force-kills the process independently of Java implementation.
+ *
+ * In Java 7, destroy() would force kill the process.
+ *
+ * Java 8 changed the behavior of destroy() to be normal
+ * termination and added destroyForcibly(), sigh.
+ *
+ * TODO: remove this class when Java 7 is no longer supported.
+ *
+ * @param process the process to destroy.
+ */
+ public static void destroyForcibly(Process process) {
+ try {
+ Method methodToFind = Process.class.getMethod("destroyForcibly", (Class<?>[]) null);
+ methodToFind.invoke(process);
+ } catch (Exception e) {
+ process.destroy();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/efa50a79/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala
index 6661816..95d01dd 100644
--- a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala
+++ b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala
@@ -25,6 +25,7 @@ import java.util.concurrent.CountDownLatch
import org.apache.samza.SamzaException
import org.apache.samza.coordinator.JobCoordinator
import org.apache.samza.job.ApplicationStatus.{New, Running, UnsuccessfulFinish}
+import org.apache.samza.job.util.ProcessKiller
import org.apache.samza.job.{ApplicationStatus, CommandBuilder, StreamJob}
import org.apache.samza.util.Logging
@@ -69,7 +70,7 @@ class ProcessJob(commandBuilder: CommandBuilder, jobCoordinator: JobCoordinator)
}
def kill: StreamJob = {
- process.destroy
+ ProcessKiller.destroyForcibly(process)
jobStatus = Some(UnsuccessfulFinish);
ProcessJob.this
}