[FLINK-14901] Throw Error in MemoryUtils if there is problem with usi…

…ng system classes over reflection

flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java

@@ -19,7 +19,6 @@
 package org.apache.flink.core.memory;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.JavaGcCleanerWrapper;
 
 import java.lang.reflect.Constructor;
@@ -49,15 +48,15 @@ private static sun.misc.Unsafe getUnsafe() {
 			unsafeField.setAccessible(true);
 			return (sun.misc.Unsafe) unsafeField.get(null);
 		} catch (SecurityException e) {
-			throw new RuntimeException("Could not access the sun.misc.Unsafe handle, permission denied by security manager.", e);
+			throw new Error("Could not access the sun.misc.Unsafe handle, permission denied by security manager.", e);
 		} catch (NoSuchFieldException e) {
-			throw new RuntimeException("The static handle field in sun.misc.Unsafe was not found.");
+			throw new Error("The static handle field in sun.misc.Unsafe was not found.");
 		} catch (IllegalArgumentException e) {
-			throw new RuntimeException("Bug: Illegal argument reflection access for static field.", e);
+			throw new Error("Bug: Illegal argument reflection access for static field.", e);
 		} catch (IllegalAccessException e) {
-			throw new RuntimeException("Access to sun.misc.Unsafe is forbidden by the runtime.", e);
+			throw new Error("Access to sun.misc.Unsafe is forbidden by the runtime.", e);
 		} catch (Throwable t) {
-			throw new RuntimeException("Unclassified error while trying to access the sun.misc.Unsafe handle.", t);
+			throw new Error("Unclassified error while trying to access the sun.misc.Unsafe handle.", t);
 		}
 	}
 
@@ -72,21 +71,20 @@ private static Constructor<? extends ByteBuffer> getDirectBufferPrivateConstruct
 			constructor.setAccessible(true);
 			return constructor;
 		} catch (NoSuchMethodException e) {
-			ExceptionUtils.rethrow(
-				e,
-				"The private constructor java.nio.DirectByteBuffer.<init>(long, int) is not available.");
+			throw new Error(
+				"The private constructor java.nio.DirectByteBuffer.<init>(long, int) is not available.",
+				e);
 		} catch (SecurityException e) {
-			ExceptionUtils.rethrow(
-				e,
+			throw new Error(
 				"The private constructor java.nio.DirectByteBuffer.<init>(long, int) is not available, " +
-					"permission denied by security manager");
+					"permission denied by security manager",
+				e);
 		} catch (Throwable t) {
-			ExceptionUtils.rethrow(
-				t,
+			throw new Error(
 				"Unclassified error while trying to access private constructor " +
-					"java.nio.DirectByteBuffer.<init>(long, int).");
+					"java.nio.DirectByteBuffer.<init>(long, int).",
+				t);
 		}
-		throw new RuntimeException("unexpected to avoid returning null");
 	}
 
 	/**
@@ -130,8 +128,7 @@ static ByteBuffer wrapUnsafeMemoryWithByteBuffer(long address, int size) {
 		try {
 			return (ByteBuffer) DIRECT_BUFFER_CONSTRUCTOR.newInstance(address, size);
 		} catch (Throwable t) {
-			ExceptionUtils.rethrow(t, "Failed to wrap unsafe off-heap memory with ByteBuffer");
+			throw new Error("Failed to wrap unsafe off-heap memory with ByteBuffer", t);
 		}
-		throw new RuntimeException("unexpected to avoid returning null");
 	}
 }

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

@@ -50,6 +50,7 @@ private static CleanerFactory findGcCleaner() {
 		CleanerFactory foundCleanerFactory = null;
 		Throwable t = null;
 		for (CleanerProvider cleanerProvider : CLEANER_PROVIDERS) {
+			//noinspection OverlyBroadCatchBlock
 			try {
 				foundCleanerFactory = cleanerProvider.createCleanerFactory();
 				break;
@@ -60,7 +61,7 @@ private static CleanerFactory findGcCleaner() {
 
 		if (foundCleanerFactory == null) {
 			String errorMessage = String.format("Failed to find GC Cleaner among available providers: %s", CLEANER_PROVIDERS);
-			throw new FlinkRuntimeException(errorMessage, t);
+			throw new Error(errorMessage, t);
 		}
 		return foundCleanerFactory;
 	}
@@ -135,7 +136,7 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
 			try {
 				cleaner = cleanerCreateMethod.invoke(null, owner, cleanupOperation);
 			} catch (IllegalAccessException | InvocationTargetException e) {
-				throw new FlinkRuntimeException("Failed to create a Java legacy Cleaner", e);
+				throw new Error("Failed to create a Java legacy Cleaner", e);
 			}
 			String ownerString = owner.toString(); // lambda should not capture the owner object
 			return () -> {
@@ -144,7 +145,7 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
 				} catch (IllegalAccessException | InvocationTargetException e) {
 					String message = String.format("FATAL UNEXPECTED - Failed to invoke a Java legacy Cleaner for %s", ownerString);
 					LOG.error(message, e);
-					throw new FlinkRuntimeException(message, e);
+					throw new Error(message, e);
 				}
 			};
 		}
@@ -238,7 +239,7 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
 			try {
 				cleanable = cleanerRegisterMethod.invoke(cleaner, owner, cleanupOperation);
 			} catch (IllegalAccessException | InvocationTargetException e) {
-				throw new FlinkRuntimeException("Failed to create a Java 9 Cleaner", e);
+				throw new Error("Failed to create a Java 9 Cleaner", e);
 			}
 			String ownerString = owner.toString(); // lambda should not capture the owner object
 			return () -> {
@@ -247,7 +248,7 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
 				} catch (IllegalAccessException | InvocationTargetException e) {
 					String message = String.format("FATAL UNEXPECTED - Failed to invoke a Java 9 Cleaner$Cleanable for %s", ownerString);
 					LOG.error(message, e);
-					throw new FlinkRuntimeException(message, e);
+					throw new Error(message, e);
 				}
 			};
 		}