[FLINK-18815] Close safety net guarded closeable iff it is still regi…

…stered

This closes apache#13124.

flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java

@@ -175,8 +175,13 @@ String getDebugString() {
 
 		@Override
 		public void close() throws IOException {
-			closeableRegistry.removeCloseableInternal(innerCloseable);
-			innerCloseable.close();
+			// Mark sure the inner closeable is still registered and thus unclosed to
+			// prevent duplicated and concurrent closing from registry closing. This could
+			// happen if registry is closing after this phantom reference was enqueued.
+			if (closeableRegistry.removeCloseableInternal(innerCloseable)) {
+				LOG.warn("Closing unclosed resource via safety-net: {}", getDebugString());
+				innerCloseable.close();
+			}
 		}
 	}
 
@@ -205,7 +210,6 @@ public void run() {
 
 					if (toClose != null) {
 						try {
-							LOG.warn("Closing unclosed resource via safety-net: {}", toClose.getDebugString());
 							toClose.close();
 						}
 						catch (Throwable t) {

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

@@ -163,9 +163,9 @@ protected final void addCloseableInternal(Closeable closeable, T metaData) {
 	/**
 	 * Removes a mapping from the registry map, respecting locking.
 	 */
-	protected final void removeCloseableInternal(Closeable closeable) {
+	protected final boolean removeCloseableInternal(Closeable closeable) {
 		synchronized (getSynchronizationLock()) {
-			closeableToRef.remove(closeable);
+			return closeableToRef.remove(closeable) != null;
 		}
 	}