Try to use jdk api to create hardlink when rename file when compaction. (#3876)

### Motivation

Current HardLink will create a process to execute mv like command to rename file in compaction logic.

maybe we can just use jdk api to do this with lower overhead.

see javadoc: https://docs.oracle.com/javase/7/docs/api/java/nio/file/Files.html#createLink(java.nio.file.Path,%20java.nio.file.Path)

### Changes

test if `Files.createLink` is available in hardlink static code block.
if test fails means `Files.createLink` is not available.

else will use `Files.createLink` when call createHardlink

(cherry picked from commit af419cc)

Author: lifepuzzlefun

Diff for: bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java

@@ -683,7 +683,7 @@ private void createNewCompactionLog() throws IOException {
     private void removeCurCompactionLog() {
         synchronized (compactionLogLock) {
             if (compactionLogChannel != null) {
-                if (!compactionLogChannel.getLogFile().delete()) {
+                if (compactionLogChannel.getLogFile().exists() && !compactionLogChannel.getLogFile().delete()) {
                     LOG.warn("Could not delete compaction log file {}", compactionLogChannel.getLogFile());
                 }
 

Diff for: bookkeeper-server/src/main/java/org/apache/bookkeeper/util/HardLink.java

@@ -22,12 +22,18 @@
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Class for creating hardlinks.
@@ -42,7 +48,7 @@
  * efficient - and minimizes the impact of the extra buffer creations.
  */
 public class HardLink {
-
+  private static final Logger LOG = LoggerFactory.getLogger(HardLink.class);
   /**
    * OS Types.
    */
@@ -395,12 +401,19 @@ protected static int getMaxAllowedCmdArgLength() {
     return getHardLinkCommand.getMaxAllowedCmdArgLength();
   }
 
+  private static final AtomicBoolean CREATE_LINK_SUPPORTED = new AtomicBoolean(true);
+
   /*
    * ****************************************************
    * Complexity is above.  User-visible functionality is below
    * ****************************************************
    */
 
+  @VisibleForTesting
+  static void enableJdkLinkApi(boolean enable) {
+    CREATE_LINK_SUPPORTED.set(enable);
+  }
+
   /**
    * Creates a hardlink.
    * @param file - existing source file
@@ -416,6 +429,23 @@ public static void createHardLink(File file, File linkName)
       throw new IOException(
           "invalid arguments to createHardLink: link name is null");
     }
+
+    // if createLink available try first, else fall back to shell command.
+    if (CREATE_LINK_SUPPORTED.get()) {
+      try {
+        Path newFile = Files.createLink(linkName.toPath(), file.toPath());
+        if (newFile.toFile().exists()) {
+          return;
+        }
+      } catch (UnsupportedOperationException e) {
+        LOG.error("createLink not supported", e);
+        CREATE_LINK_SUPPORTED.set(false);
+      } catch (IOException e) {
+        LOG.error("error when create hard link use createLink", e);
+        CREATE_LINK_SUPPORTED.set(false);
+      }
+    }
+
     // construct and execute shell command
     String[] hardLinkCommand = getHardLinkCommand.linkOne(file, linkName);
     Process process = Runtime.getRuntime().exec(hardLinkCommand);

Diff for: bookkeeper-server/src/test/java/org/apache/bookkeeper/util/TestHardLink.java

@@ -0,0 +1,82 @@
+/*
+ * 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.bookkeeper.util;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+import org.apache.commons.io.FileUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+
+public class TestHardLink {
+
+    private File tempDir;
+
+    @Before
+    public void setup() throws IOException {
+        // Create at least one file so that target disk will never be empty
+        tempDir = IOUtils.createTempDir("TestHardLink", "test-hardlink");
+    }
+
+    @After
+    public void tearDown() throws IOException {
+        FileUtils.deleteDirectory(tempDir);
+    }
+
+    private void verifyHardLink(File origin, File linkedOrigin) throws IOException {
+        Assert.assertTrue(origin.exists());
+        Assert.assertFalse(linkedOrigin.exists());
+
+        HardLink.createHardLink(origin, linkedOrigin);
+
+        Assert.assertTrue(origin.exists());
+        Assert.assertTrue(linkedOrigin.exists());
+
+        // when delete origin file it should be success and not exist.
+        origin.delete();
+        Assert.assertFalse(origin.exists());
+        Assert.assertTrue(linkedOrigin.exists());
+    }
+
+    @Test
+    public void testHardLink() throws IOException {
+        String uuidSuffix = UUID.randomUUID().toString();
+
+        // prepare file
+        File origin = new File(tempDir, "originFile." + uuidSuffix);
+        File linkedOrigin = new File(tempDir, "linkedOrigin." + uuidSuffix);
+        origin.createNewFile();
+
+        // disable jdk api link first
+        HardLink.enableJdkLinkApi(false);
+        verifyHardLink(origin, linkedOrigin);
+
+        // prepare file
+        File jdkorigin = new File(tempDir, "jdkoriginFile." + uuidSuffix);
+        File jdklinkedOrigin = new File(tempDir, "jdklinkedOrigin." + uuidSuffix);
+        jdkorigin.createNewFile();
+
+        // enable jdk api link
+        HardLink.enableJdkLinkApi(true);
+        verifyHardLink(jdkorigin, jdklinkedOrigin);
+    }
+}