[helix] branch master updated: Change thread pool used for TTL-based GC in ZkBucketDataAccessor (#945)

[hulee]

This is an automated email from the ASF dual-hosted git repository.

hulee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git


The following commit(s) were added to refs/heads/master by this push:
     new 2e9ac43  Change thread pool used for TTL-based GC in 
ZkBucketDataAccessor (#945)
2e9ac43 is described below

commit 2e9ac43cc0d76f5ef2dd408ec506a4ac5f4e5b1c
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Fri Apr 10 15:40:03 2020 -0700

    Change thread pool used for TTL-based GC in ZkBucketDataAccessor (#945)
    
    ZkBucketDataAccessor supports TTL-based garbage collection of stale 
versions. We create a thread pool for this of size 1, but the way the thread 
pool is created currently may not guarantee strictly sequential execution of GC 
timer tasks, which may cause unexpected GC behaviors. This commit uses 
Executors.newSingleThreadScheduledExecutor()to create the GC thread pool with a 
sequential execution guarantee.
    
    Note: another way to do this is to implement a Bounded event queue-based 
GC, but Java already provides this queue-based model in Executors interface - 
we want to try our best at avoid reinventing the wheel.
---
 .../helix/manager/zk/ZkBucketDataAccessor.java     | 23 ++++++++++++----------
 1 file changed, 13 insertions(+), 10 deletions(-)

diff --git 
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
 
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
index 1ebab28..f61fb50 100644
--- 
a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
+++ 
b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
@@ -36,7 +36,7 @@ import org.apache.helix.HelixException;
 import org.apache.helix.HelixProperty;
 import org.apache.helix.SystemPropertyKeys;
 import org.apache.helix.msdcommon.exception.InvalidRoutingDataException;
-import org.apache.helix.util.GZipCompressionUtil;
+import org.apache.helix.zookeeper.util.GZipCompressionUtil;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
@@ -62,7 +62,10 @@ public class ZkBucketDataAccessor implements 
BucketDataAccessor, AutoCloseable {
   private static final String LAST_WRITE_KEY = "LAST_WRITE";
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
   // Thread pool for deleting stale versions
-  private static final ScheduledExecutorService GC_THREAD = 
Executors.newScheduledThreadPool(1);
+  // Note that newScheduledThreadPool(1) may not work. 
newSingleThreadScheduledExecutor guarantees
+  // sequential execution, which is what we want for implementing TTL & GC here
+  private static final ScheduledExecutorService GC_THREAD =
+      Executors.newSingleThreadScheduledExecutor();
 
   private final int _bucketSize;
   private final long _versionTTL;
@@ -359,26 +362,26 @@ public class ZkBucketDataAccessor implements 
BucketDataAccessor, AutoCloseable {
 
   /**
    * Filter out non-version children names and non-stale versions.
-   * @param children
+   * @param childrenToRemove
    */
-  private void filterChildrenNames(List<String> children, String 
currentVersion) {
+  private void filterChildrenNames(List<String> childrenToRemove, String 
currentVersion) {
     // Leave out metadata
-    children.remove(LAST_SUCCESSFUL_WRITE_KEY);
-    children.remove(LAST_WRITE_KEY);
+    childrenToRemove.remove(LAST_SUCCESSFUL_WRITE_KEY);
+    childrenToRemove.remove(LAST_WRITE_KEY);
 
     // Leave out currentVersion and above
     // This is because we want to honor the TTL for newer versions
-    children.remove(currentVersion);
+    childrenToRemove.remove(currentVersion);
     long currentVer = Long.parseLong(currentVersion);
-    for (String child : children) {
+    for (String child : childrenToRemove) {
       try {
         long version = Long.parseLong(child);
         if (version >= currentVer) {
-          children.remove(child);
+          childrenToRemove.remove(child);
         }
       } catch (Exception e) {
         // Ignore ZNode names that aren't parseable
-        children.remove(child);
+        childrenToRemove.remove(child);
         LOG.debug("Found an invalid ZNode: {}", child);
       }
     }