Repository: apache/incubator-heron (language: python)

Issue:
Uploader implementations may leak storage when resubmitting the same analytic multiple times
**Describe the bug**
The Uploader logic in Heron adds a Random Long to the filename that causes memory leak in Bookkeeper.
The Uploader filename generation logic is [here](https://github.com/apache/incubator-heron/blob/abb2767e3df3ca6eba009f46efe1f1e83695617a/heron/spi/src/java/org/apache/heron/spi/utils/UploaderUtils.java#L62).

And the Dlog config that disables time and size based retention is [here](https://github.com/apache/incubator-heron/blob/abb2767e3df3ca6eba009f46efe1f1e83695617a/heron/uploaders/src/java/org/apache/heron/uploader/dlog/DLUploader.java#L122-L128).

Eventually the Bookkeeper log runs out of space and no more topologies can be submitted.

The proposed fix is to remove the Random.long that is being added to the filename.


Discussion / hints:
Would using [RandomStringUtils#random](https://commons.apache.org/proper/commons-lang/apidocs/org/apache/commons/lang3/RandomStringUtils.html) with a character count of 11 fix the issue? Or is this a string length related issue?

If my math is correct it should generate 26 `uppercase` + 26 `lowercase` + 10 `digits` permute 11 `count` = 62 p 11 = 390,164,706,723,052,800 potential strings. That is more options than `Random#long` provides.
So my suggestion is to remove the Random.long all together. I'm not sure why this was added, but we don't have any ability to "roll-back" to the previous uploads. So having an endless series of entries in Bookkeeper doesn't make sense. If I resubmit the same topology with name, role, version coordinates, I would expect it to replace the previous entry in Bookkeeper (or whatever upload mechanism is used). For example when developing and testing an analytic job.

If the desire is to run different binaries, one should distinguish the name, role and/or version to create a unique coordinate in the Upload/Download mechanism. 

We will have to test the Upload functionality to ensure that removing the Random.long doesn't break anything.
Is that possible to, either defined by user, or set it with timestamp instead?

Fix patch:
diff --git a/deploy/kubernetes/helm/templates/bookie.yaml b/deploy/kubernetes/helm/templates/bookie.yaml
index 9cc62c81bde..9e6e812feb3 100644
--- a/deploy/kubernetes/helm/templates/bookie.yaml
+++ b/deploy/kubernetes/helm/templates/bookie.yaml
@@ -43,9 +43,6 @@ data:
   BK_indexDirectories: "/bookkeeper/data/ledgers" 
   BK_zkServers: {{ .Release.Name }}-zookeeper:{{ .Values.zookeeper.clientPort }}
   BK_autoRecoveryDaemonEnabled: "true"
-  BK_journalMaxBackups: "{{ .Values.bookkeeper.journal.maxBackups }}"
-  BK_journalMaxSizeMB: "{{ .Values.bookkeeper.journal.maxSizeMB }}"
-  BK_logSizeLimit: "{{ int64 .Values.bookkeeper.logSizeLimit }}"
   {{- if .Values.bookkeeper.useHostNameAsBookieID }}
   BK_useHostNameAsBookieID: "true"
   {{- end }}
diff --git a/deploy/kubernetes/helm/values.yaml.template b/deploy/kubernetes/helm/values.yaml.template
index e02797d40dd..9ffbad7a739 100644
--- a/deploy/kubernetes/helm/values.yaml.template
+++ b/deploy/kubernetes/helm/values.yaml.template
@@ -89,8 +89,6 @@ bookkeeper:
   useStatefulSet: true
   affinityPods: false
   useVolumeClaimTemplate: true
-
-  logSizeLimit: 10000000
   useHostNameAsBookieID: true
   tolerateUnreadyEndpoints: true
 
@@ -108,8 +106,6 @@ bookkeeper:
   useHostPath: false
   journal:
     capacity: 5G
-    maxBackups: 3
-    maxSizeMB: 300
     hostPath: '/opt/bookkeeper/joural-disk'
   ledgers:
     capacity: 15G
diff --git a/heron/spi/src/java/org/apache/heron/spi/utils/UploaderUtils.java b/heron/spi/src/java/org/apache/heron/spi/utils/UploaderUtils.java
index 4365f5a715d..53d16460200 100644
--- a/heron/spi/src/java/org/apache/heron/spi/utils/UploaderUtils.java
+++ b/heron/spi/src/java/org/apache/heron/spi/utils/UploaderUtils.java
@@ -23,7 +23,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.util.Random;
 
 /**
  * Utility used by Uploader
@@ -65,8 +64,8 @@ public static String generateFilename(
       String tag,
       int version,
       String extension) {
-    return String.format("%s-%s-%s-%d-%d%s",
-        topologyName, role, tag, version, new Random().nextLong(), extension);
+    return String.format("%s-%s-%s-%d%s",
+        topologyName, role, tag, version, extension);
   }
 
   public static void copyToOutputStream(String inFile,


Test patch:
diff --git a/heron/spi/tests/java/org/apache/heron/spi/utils/UploaderUtilsTest.java b/heron/spi/tests/java/org/apache/heron/spi/utils/UploaderUtilsTest.java
index f7031b0e870..08958b1388c 100644
--- a/heron/spi/tests/java/org/apache/heron/spi/utils/UploaderUtilsTest.java
+++ b/heron/spi/tests/java/org/apache/heron/spi/utils/UploaderUtilsTest.java
@@ -37,19 +37,26 @@ public class UploaderUtilsTest {
 
   @Test
   public void testGenerateFilename() throws Exception {
-    int expectedUniqueFilename = 10000;
+    int requestsForFilename = 2;
     String topologyName = "topologyName";
     String role = "role";
     String tag = "";
     int version = -1;
     Set<String> filenames = new HashSet<>();
-    for (int i = 0; i < expectedUniqueFilename; i++) {
+    for (int i = 0; i < requestsForFilename; i++) {
       filenames.add(UploaderUtils.generateFilename(
           topologyName, role, tag, version, ""));
     }
 
-    // All filenames should be unique
-    Assert.assertEquals(expectedUniqueFilename, filenames.size());
+    // There should not be multiple entries
+    Assert.assertEquals(1, filenames.size());
+
+    int newVersion = 1;
+    filenames.add(UploaderUtils.generateFilename(
+        topologyName, role, tag, newVersion, ""));
+
+    // Each version should provide a unique filename
+    Assert.assertEquals(2, filenames.size());
   }
 
   @Test