please review #3224
 Fixed by #3224

`file = null;` in the `finally` block is evaluated even when `file` is `null`.  I would move the null-check out of the try block and return early when `file` is `null`.

Support FileRegion without pre-initialized FileDescriptor
DefaultFileRegion requires a FileDescriptor in its constructor, which means we need to have an opened file handle. In super large workloads where we queue up a lot of file regions, this could lead to too many open files due to the way these file descriptors are cleaned.

In Spark we created a LazyFileRegion implementation (see https://github.com/apache/spark/pull/3172), but that wouldn't work with the native epoll transport. Would be great to push that into Netty and enable support in native transport.

cc @normanmaurer 

transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java transport/src/main/java/io/netty/channel/DefaultFileRegion.java
transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java transport/src/main/java/io/netty/channel/DefaultFileRegion.java
diff --git a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
index 2a06c2267cb..24092d03c9e 100644
--- a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
+++ b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.c
@@ -1101,7 +1101,7 @@ JNIEXPORT jint JNICALL Java_io_netty_channel_epoll_Native_accept(JNIEnv * env, j
     return socketFd;
 }
 
-JNIEXPORT jlong JNICALL Java_io_netty_channel_epoll_Native_sendfile(JNIEnv *env, jclass clazz, jint fd, jobject fileRegion, jlong base_off, jlong off, jlong len) {
+JNIEXPORT jlong JNICALL Java_io_netty_channel_epoll_Native_sendfile0(JNIEnv *env, jclass clazz, jint fd, jobject fileRegion, jlong base_off, jlong off, jlong len) {
     jobject fileChannel = (*env)->GetObjectField(env, fileRegion, fileChannelFieldId);
     if (fileChannel == NULL) {
         throwRuntimeException(env, "Unable to obtain FileChannel from FileRegion");
diff --git a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
index a6e9fb04aae..92f6ee627bb 100644
--- a/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
+++ b/transport-native-epoll/src/main/c/io_netty_channel_epoll_Native.h
@@ -69,7 +69,7 @@ void Java_io_netty_channel_epoll_Native_listen(JNIEnv * env, jclass clazz, jint
 jboolean Java_io_netty_channel_epoll_Native_connect(JNIEnv * env, jclass clazz, jint fd, jbyteArray address, jint scopeId, jint port);
 jboolean Java_io_netty_channel_epoll_Native_finishConnect(JNIEnv * env, jclass clazz, jint fd);
 jint Java_io_netty_channel_epoll_Native_accept(JNIEnv * env, jclass clazz, jint fd);
-jlong Java_io_netty_channel_epoll_Native_sendfile(JNIEnv *env, jclass clazz, jint fd, jobject fileRegion, jlong base_off, jlong off, jlong len);
+jlong Java_io_netty_channel_epoll_Native_sendfile0(JNIEnv *env, jclass clazz, jint fd, jobject fileRegion, jlong base_off, jlong off, jlong len);
 jobject Java_io_netty_channel_epoll_Native_remoteAddress(JNIEnv * env, jclass clazz, jint fd);
 jobject Java_io_netty_channel_epoll_Native_localAddress(JNIEnv * env, jclass clazz, jint fd);
 void Java_io_netty_channel_epoll_Native_setReuseAddress(JNIEnv * env, jclass clazz, jint fd, jint optval);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
index 5bbd890692b..9b6894ee4fc 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
@@ -77,7 +77,16 @@ public static native long writevAddresses(int fd, long memoryAddress, int length
     public static native int read(int fd, ByteBuffer buf, int pos, int limit) throws IOException;
     public static native int readAddress(int fd, long address, int pos, int limit) throws IOException;
 
-    public static native long sendfile(
+    public static long sendfile(
+            int dest, DefaultFileRegion src, long baseOffset, long offset, long length) throws IOException {
+        // Open the file-region as it may be created via the lazy constructor. This is needed as we directly access
+        // the FileChannel field directly via JNI
+        src.open();
+
+        return sendfile0(dest, src, baseOffset, offset, length);
+    }
+
+    private static native long sendfile0(
             int dest, DefaultFileRegion src, long baseOffset, long offset, long length) throws IOException;
 
     public static int sendTo(
diff --git a/transport/src/main/java/io/netty/channel/DefaultFileRegion.java b/transport/src/main/java/io/netty/channel/DefaultFileRegion.java
index c2ac264cc61..dae94e45c28 100644
--- a/transport/src/main/java/io/netty/channel/DefaultFileRegion.java
+++ b/transport/src/main/java/io/netty/channel/DefaultFileRegion.java
@@ -16,15 +16,18 @@
 package io.netty.channel;
 
 import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.IllegalReferenceCountException;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
+import java.io.File;
 import java.io.IOException;
+import java.io.RandomAccessFile;
 import java.nio.channels.FileChannel;
 import java.nio.channels.WritableByteChannel;
 
 /**
- * Default {@link FileRegion} implementation which transfer data from a {@link FileChannel}.
+ * Default {@link FileRegion} implementation which transfer data from a {@link FileChannel} or {@link File}.
  *
  * Be aware that the {@link FileChannel} will be automatically closed once {@link #refCnt()} returns
  * {@code 0}.
@@ -32,11 +35,11 @@
 public class DefaultFileRegion extends AbstractReferenceCounted implements FileRegion {
 
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultFileRegion.class);
-
-    private final FileChannel file;
+    private final File f;
     private final long position;
     private final long count;
     private long transfered;
+    private FileChannel file;
 
     /**
      * Create a new instance
@@ -58,6 +61,47 @@ public DefaultFileRegion(FileChannel file, long position, long count) {
         this.file = file;
         this.position = position;
         this.count = count;
+        f = null;
+    }
+
+    /**
+     * Create a new instance using the given {@link File}. The {@link File} will be opened lazily or
+     * explicitly via {@link #open()}.
+     *
+     * @param f         the {@link File} which should be transfered
+     * @param position  the position from which the transfer should start
+     * @param count     the number of bytes to transfer
+     */
+    public DefaultFileRegion(File f, long position, long count) {
+        if (f == null) {
+            throw new NullPointerException("f");
+        }
+        if (position < 0) {
+            throw new IllegalArgumentException("position must be >= 0 but was " + position);
+        }
+        if (count < 0) {
+            throw new IllegalArgumentException("count must be >= 0 but was " + count);
+        }
+        this.position = position;
+        this.count = count;
+        this.f = f;
+    }
+
+    /**
+     * Returns {@code true} if the {@link FileRegion} has a open file-descriptor
+     */
+    public boolean isOpen() {
+        return file != null;
+    }
+
+    /**
+     * Explicitly open the underlying file-descriptor if not done yet.
+     */
+    public void open() throws IOException {
+        if (!isOpen() && refCnt() > 0) {
+            // Only open if this DefaultFileRegion was not released yet.
+            file = new RandomAccessFile(f, "r").getChannel();
+        }
     }
 
     @Override
@@ -86,6 +130,11 @@ public long transferTo(WritableByteChannel target, long position) throws IOExcep
         if (count == 0) {
             return 0L;
         }
+        if (refCnt() == 0) {
+            throw new IllegalReferenceCountException(0);
+        }
+        // Call open to make sure fc is initialized. This is a no-oop if we called it before.
+        open();
 
         long written = file.transferTo(this.position + position, count, target);
         if (written > 0) {
@@ -97,11 +146,15 @@ public long transferTo(WritableByteChannel target, long position) throws IOExcep
     @Override
     protected void deallocate() {
         try {
-            file.close();
+            if (file != null) {
+                file.close();
+            }
         } catch (IOException e) {
             if (logger.isWarnEnabled()) {
                 logger.warn("Failed to close a file.", e);
             }
+        } finally {
+            file = null;
         }
     }
 }
