[FLINK-3438] ExternalProcessRunner fails to detect ClassNotFound exception...

[FLINK-3438] ExternalProcessRunner fails to detect ClassNotFound exception because of locale settings

[FLINK-3438] Improved solution, no workaround

[FLINK-3438] Change: a faulty process now causes a RuntimeException to be thrown

This closes #1665.

flink-core/src/main/java/org/apache/flink/util/ExternalProcessRunner.java

@@ -40,8 +40,6 @@ import java.util.List;
  */
 @Internal
 public class ExternalProcessRunner {
-	private final String entryPointClassName;
-
 	private final Process process;
 
 	private final Thread pipeForwarder;
@@ -53,8 +51,6 @@ public class ExternalProcessRunner {
 	 * The class must have a "main" method.
 	 */
 	public ExternalProcessRunner(String entryPointClassName, String[] parameters) throws IOException {
-		this.entryPointClassName = entryPointClassName;
-
 		String javaCommand = getJavaCommandPath();
 
 		List<String> commandList = new ArrayList<>();
@@ -92,9 +88,10 @@ public class ExternalProcessRunner {
 			pipeForwarder.join();
 
 			if (returnCode != 0) {
-				// determine whether we failed because of a ClassNotFoundException and forward that
-				if (getErrorOutput().toString().contains("Error: Could not find or load main class " + entryPointClassName)) {
-					throw new ClassNotFoundException("Error: Could not find or load main class " + entryPointClassName);
+
+				final String errorOutput = getErrorOutput().toString();
+				if (!errorOutput.isEmpty()) {
+					throw new RuntimeException(errorOutput);
 				}
 
 			}

flink-core/src/test/java/org/apache/flink/util/ExternalProcessRunnerTest.java

@@ -20,15 +20,21 @@ package org.apache.flink.util;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 public class ExternalProcessRunnerTest {
 
 	@Test(expected = ClassNotFoundException.class)
 	public void testClassNotFound() throws Exception {
-		ExternalProcessRunner runner = new ExternalProcessRunner("MyClassThatDoesNotExist", new String[]{});
-		runner.run();
+		final String nonExistingClassName = "MyClassThatDoesNotExist";
+		ExternalProcessRunner runner = new ExternalProcessRunner(nonExistingClassName, new String[]{});
+		try {
+			runner.run();
+		} catch (final Exception e) {
+			if (e.getMessage().contains(nonExistingClassName)) {
+				throw new ClassNotFoundException();
+			}
+		}
 	}
 
 	@Test
@@ -64,18 +70,12 @@ public class ExternalProcessRunnerTest {
 		assertEquals(runner.getErrorOutput().toString(), "Hello process hello42" + System.lineSeparator());
 	}
 
-	@Test
+	@Test(expected = RuntimeException.class)
 	public void testFailing() throws Exception {
 		final ExternalProcessRunner runner = new ExternalProcessRunner(Failing.class.getName(), new String[]{});
-
-		int result = runner.run();
-
-		assertEquals(1, result);
-		// this needs to be adapted if the test changes because it contains the line number
-		assertTrue(runner.getErrorOutput().toString().startsWith("Exception in thread \"main\""));
+		runner.run();
 	}
 
-
 	public static class InfiniteLoop {
 		public static void main(String[] args) {
 			while (true) {