[
https://issues.apache.org/jira/browse/FLINK-4910?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15611883#comment-15611883
]
ASF GitHub Bot commented on FLINK-4910:
---------------------------------------
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85335033
--- Diff:
flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java
---
@@ -0,0 +1,165 @@
+/*
+ * 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.flink.core.fs;
+
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This implementation of an {@link AbstractCloseableRegistry} registers
{@link WrappingProxyCloseable}. When
+ * the proxy becomes subject to GC, this registry takes care of closing
unclosed {@link Closeable}s.
+ * <p>
+ * Phantom references are used to track when {@link
org.apache.flink.util.WrappingProxy}s of {@link Closeable} got
+ * GC'ed. We ensure that the wrapped {@link Closeable} is proeprly closed
to avoid resource leaks.
+ * <p>
+ * Other than that, it works like a normal {@link CloseableRegistry}.
+ * <p>
+ * All methods in this class are thread-safe.
+ */
+public class SafetyNetCloseableRegistry extends
+ AbstractCloseableRegistry<WrappingProxyCloseable<? extends
Closeable>,
+
SafetyNetCloseableRegistry.PhantomDelegatingCloseableRef> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SafetyNetCloseableRegistry.class);
+ private final ReferenceQueue<WrappingProxyCloseable<? extends
Closeable>> referenceQueue;
+ private final Thread reaperThread;
+
+ public SafetyNetCloseableRegistry() {
+ super(new IdentityHashMap<Closeable,
PhantomDelegatingCloseableRef>());
+ this.referenceQueue = new ReferenceQueue<>();
+ this.reaperThread = new Thread(new CloseableReaperRunnable());
+ reaperThread.start();
+ }
+
+ @Override
+ protected void doRegister(
+ WrappingProxyCloseable<? extends Closeable>
wrappingProxyCloseable,
+ Map<Closeable, PhantomDelegatingCloseableRef>
closeableMap) throws IOException {
+
+ Closeable innerCloseable =
WrappingProxyUtil.stripProxy(wrappingProxyCloseable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ PhantomDelegatingCloseableRef phantomRef =
+ new
PhantomDelegatingCloseableRef(wrappingProxyCloseable, referenceQueue);
+
+ closeableMap.put(innerCloseable, phantomRef);
+ }
+
+ @Override
+ protected void doUnRegister(
+ WrappingProxyCloseable<? extends Closeable> closeable,
+ Map<Closeable, PhantomDelegatingCloseableRef>
closeableMap) {
+
+ Closeable innerCloseable =
WrappingProxyUtil.stripProxy(closeable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ closeableMap.remove(innerCloseable);
+ }
+
+ /**
+ * Phantom reference to {@link WrappingProxyCloseable}.
+ */
+ static final class PhantomDelegatingCloseableRef
+ extends PhantomReference<WrappingProxyCloseable<?
extends Closeable>>
+ implements Closeable {
+
+ private final Closeable innerCloseable;
+ private final String debugString;
+
+ public PhantomDelegatingCloseableRef(
+ WrappingProxyCloseable<? extends Closeable>
referent,
+ ReferenceQueue<? super WrappingProxyCloseable<?
extends Closeable>> q) {
+
+ super(referent, q);
+ this.innerCloseable =
Preconditions.checkNotNull(WrappingProxyUtil.stripProxy(referent));
+ this.debugString = referent.toString();
+ }
+
+ public Closeable getInnerCloseable() {
+ return innerCloseable;
+ }
+
+ public String getDebugString() {
+ return debugString;
+ }
+
+ @Override
+ public void close() throws IOException {
+ innerCloseable.close();
+ }
+ }
+
+ /**
+ * Reaper runnable collects and closes leaking resources
+ */
+ final class CloseableReaperRunnable implements Runnable {
+ @Override
+ public void run() {
+ try {
+ List<PhantomDelegatingCloseableRef>
closeableList = new LinkedList<>();
--- End diff --
Isn't an array list more efficient here, especially if you reuse the list?
> Introduce safety net for closing file system streams
> ----------------------------------------------------
>
> Key: FLINK-4910
> URL: https://issues.apache.org/jira/browse/FLINK-4910
> Project: Flink
> Issue Type: Improvement
> Reporter: Stefan Richter
> Assignee: Stefan Richter
>
> Streams that are opened through {{FileSystem}} must be closed at the end of
> their life cycle. However, we found hints that some code forgets to close
> such streams.
> We should introduce i) a mechanism that closes leaking unclosed streams after
> usage and ii) provides logging that helps us to track down and fi the sources
> of such leaks.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
