narendly commented on a change in pull request #506: Increase parallelism for
ZkBucketDataAccessor
URL: https://github.com/apache/helix/pull/506#discussion_r336716846
##########
File path:
helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
##########
@@ -19,56 +19,60 @@
* under the License.
*/
+import com.google.common.collect.ImmutableMap;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
+import java.util.TimerTask;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.I0Itec.zkclient.DataUpdater;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.helix.AccessOption;
-import org.apache.helix.BaseDataAccessor;
import org.apache.helix.BucketDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixProperty;
import org.apache.helix.ZNRecord;
import org.apache.helix.manager.zk.client.DedicatedZkClientFactory;
import org.apache.helix.manager.zk.client.HelixZkClient;
-import org.apache.helix.manager.zk.client.SharedZkClientFactory;
import org.apache.helix.util.GZipCompressionUtil;
+import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ZkBucketDataAccessor implements BucketDataAccessor, AutoCloseable
{
- private static Logger LOG =
LoggerFactory.getLogger(ZkBucketDataAccessor.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(ZkBucketDataAccessor.class);
- private static final int DEFAULT_NUM_VERSIONS = 2;
+ private static final int DEFAULT_BUCKET_SIZE = 50 * 1024; // 50KB
+ private static final long DEFAULT_VERSION_TTL =
TimeUnit.MINUTES.toMillis(1L); // 1 min
private static final String BUCKET_SIZE_KEY = "BUCKET_SIZE";
private static final String DATA_SIZE_KEY = "DATA_SIZE";
- private static final String WRITE_LOCK_KEY = "WRITE_LOCK";
- private static final String LAST_SUCCESS_KEY = "LAST_SUCCESS";
+ private static final String METADATA_KEY = "METADATA";
+ private static final String LAST_SUCCESSFUL_WRITE_KEY =
"LAST_SUCCESSFUL_WRITE";
+ 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);
- // 100 KB for default bucket size
- private static final int DEFAULT_BUCKET_SIZE = 50 * 1024;
private final int _bucketSize;
- private final int _numVersions;
+ private final long _versionTTL;
private ZkSerializer _zkSerializer;
private HelixZkClient _zkClient;
- private HelixZkClient _znRecordClient;
- private BaseDataAccessor _zkBaseDataAccessor;
- private BaseDataAccessor<ZNRecord> _znRecordBaseDataAccessor;
+ private ZkBaseDataAccessor<byte[]> _zkBaseDataAccessor;
/**
* Constructor that allows a custom bucket size.
* @param zkAddr
* @param bucketSize
- * @param numVersions number of versions to store in ZK
+ * @param versionTTL in ms
Review comment:
Changed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@helix.apache.org
For additional commands, e-mail: reviews-h...@helix.apache.org
