Clean the local cache of downloaded packages if it already exists (ap…

…ache#1674)
teazj · Apr 27, 2018 · 070a9f6 · 070a9f6
1 parent f46d8eb
commit 070a9f6
Showing 1 changed file with 25 additions and 22 deletions.
diff --git a/...r-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/...r-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -117,41 +117,44 @@ private void startFunction(FunctionRuntimeInfo functionRuntimeInfo) throws Excep
             pkgDir,
             new File(FunctionDetailsUtils.getDownloadFileName(functionMetaData.getFunctionDetails())).getName());
 
-        if (!pkgFile.exists()) {
-            // download only when the package file doesn't exist
-            File tempPkgFile;
-            while (true) {
-                tempPkgFile = new File(
+        if (pkgFile.exists()) {
+            log.warn("Function package exists already {} deleting it",
+                    pkgFile);
+            pkgFile.delete();
+        }
+
+        File tempPkgFile;
+        while (true) {
+            tempPkgFile = new File(
                     pkgDir,
                     pkgFile.getName() + "." + instanceId + "." + UUID.randomUUID().toString());
-                if (!tempPkgFile.exists() && tempPkgFile.createNewFile()) {
-                    break;
-                }
+            if (!tempPkgFile.exists() && tempPkgFile.createNewFile()) {
+                break;
             }
-            try {
-                log.info("Function package file {} will be downloaded from {}",
+        }
+        try {
+            log.info("Function package file {} will be downloaded from {}",
                     tempPkgFile, functionMetaData.getPackageLocation());
-                Utils.downloadFromBookkeeper(
+            Utils.downloadFromBookkeeper(
                     dlogNamespace,
                     new FileOutputStream(tempPkgFile),
                     functionMetaData.getPackageLocation().getPackagePath());
 
-                // create a hardlink, if there are two concurrent createLink operations, one will fail.
-                // this ensures one instance will successfully download the package.
-                try {
-                    Files.createLink(
+            // create a hardlink, if there are two concurrent createLink operations, one will fail.
+            // this ensures one instance will successfully download the package.
+            try {
+                Files.createLink(
                         Paths.get(pkgFile.toURI()),
                         Paths.get(tempPkgFile.toURI()));
-                    log.info("Function package file is linked from {} to {}",
+                log.info("Function package file is linked from {} to {}",
                         tempPkgFile, pkgFile);
-                } catch (FileAlreadyExistsException faee) {
-                    // file already exists
-                    log.warn("Function package has been downloaded from {} and saved at {}",
+            } catch (FileAlreadyExistsException faee) {
+                // file already exists
+                log.warn("Function package has been downloaded from {} and saved at {}",
                         functionMetaData.getPackageLocation(), pkgFile);
-                }
-            } finally {
-                tempPkgFile.delete();
             }
+        } finally {
+            tempPkgFile.delete();
         }
 
         InstanceConfig instanceConfig = new InstanceConfig();