You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "lnbest0707-uber (via GitHub)" <gi...@apache.org> on 2024/12/19 01:00:36 UTC
[PR] Add predownload functionality to Pinot [pinot]
lnbest0707-uber opened a new pull request, #14686:
URL: https://github.com/apache/pinot/pull/14686
`feature``performance`
https://github.com/apache/pinot/issues/14592
Add a pre-download feature to enable "graceful" node replacement on Pinot. With this feature, during node replacement, admins would replace the old node (OD) to new node (NN) with the same instance id. Instead of bringing down the ON before starting up NN (which is required because there cannot be 2 nodes taking same helix id at the same time), admins could:
1. Start NN in "pre-download" mode by adding one more parameter to StartServerCommand
2. Waiting for NN "pre-download" complete with one of following conditions:
- pre-download fully succeed
- pre-download partially succeed but have retried enough times
- pre-download failed in non-retriable mode
- already waited for a max time period
3. Bring down the ON
4. Start NN in normal mode
With a successful pre-download, the Pinot node replacement could perform in the same way as a node restart. We could observe the downtime (referring to helix pending message values decrease to 0) reduce significantly.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934488467
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ @SuppressWarnings("NullAway.Init")
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _started;
+
+ public ZKClient(String zkAddress, String clusterName, String instanceName) {
Review Comment:
Good idea! Done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978434991
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,430 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ // TODO: make download timeout configurable
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ private PredownloadZKClient _predownloadZkClient;
+ private List<PredownloadSegmentInfo> _predownloadSegmentInfoList;
+ private Map<String, PredownloadTableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompletionReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_predownloadZkClient != null) {
+ _predownloadZkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _predownloadZkClient = new PredownloadZKClient(_zkAddress, _clusterName, _instanceId);
+ _predownloadZkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ PredownloadStatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _predownloadSegmentInfoList = _predownloadZkClient.getSegmentsOfInstance(_predownloadZkClient.getDataAccessor());
+ if (_predownloadSegmentInfoList.isEmpty()) {
+ PredownloadCompletionReason reason = PredownloadCompletionReason.NO_SEGMENT_TO_PREDOWNLOAD;
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, "");
+ }
+ _tableInfoMap = new HashMap<>();
+ _predownloadZkClient.updateSegmentMetadata(_predownloadSegmentInfoList, _tableInfoMap,
+ _instanceDataManagerConfig);
+ }
+
+ @VisibleForTesting
+ void loadSegmentsFromLocal() {
+ LOGGER.info("Loading segments from local to reduce number of segments to download");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (PredownloadSegmentInfo predownloadSegmentInfo : _predownloadSegmentInfoList) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ boolean loadSegmentSuccess = false;
+ try {
+ PredownloadTableInfo predownloadTableInfo = _tableInfoMap.get(predownloadSegmentInfo.getTableNameWithType());
+ if (predownloadTableInfo != null) {
+ loadSegmentSuccess =
+ predownloadTableInfo.loadSegmentFromLocal(predownloadSegmentInfo, _instanceDataManagerConfig);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to load from local for segment: {} of table: {} with issue ",
+ predownloadSegmentInfo.getSegmentName(), predownloadSegmentInfo.getTableNameWithType(), e);
Review Comment:
can we reuse `predownloadSegmentInfo.getTableNameWithType()`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934426279
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompleteReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _zkClient = new ZKClient(_zkAddress, _clusterName, _instanceId);
+ _zkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ StatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _segmentInfoList = _zkClient.getSegmentsOfInstance(_zkClient.getDataAccessor());
+ if (_segmentInfoList.isEmpty()) {
+ PredownloadCompleteReason reason = PredownloadCompleteReason.NO_SEGMENT_TO_PREDOWNLOAD;
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, "");
+ }
+ _tableInfoMap = new HashMap<>();
+ _zkClient.updateSegmentMetadata(_segmentInfoList, _tableInfoMap, _instanceDataManagerConfig);
+ }
+
+ @VisibleForTesting
+ void loadSegmentsFromLocal() {
+ LOGGER.info("Loading segments from local to reduce number of segments to download");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (SegmentInfo segmentInfo : _segmentInfoList) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ boolean loadSegmentSuccess = false;
+ try {
+ TableInfo tableInfo = _tableInfoMap.get(segmentInfo.getTableNameWithType());
+ if (tableInfo != null) {
+ loadSegmentSuccess = tableInfo.loadSegmentFromLocal(segmentInfo, _instanceDataManagerConfig);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to load from local for segment: {} of table: {} with issue ",
+ segmentInfo.getSegmentName(), segmentInfo.getTableNameWithType(), e);
+ }
+ if (!loadSegmentSuccess && segmentInfo.canBeDownloaded()) {
+ _failedSegments.add(segmentInfo.getSegmentName());
+ }
+ }, _executor);
+
+ futures.add(future);
+ }
+
+ // Wait for all CompletableFuture tasks to complete or timeout
+ CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+
+ try {
+ // Wait indefinitely for all tasks to complete
+ allOf.get(LOAD_SEGMENTS_TIMEOUT_MIN, TimeUnit.MINUTES);
Review Comment:
To ensure there's only one pre-download process running on the host/container, the responsibility is on the spawning process.
Within this process, the `start` method which triggers segments loading and downloading is running in synchronous mode. The source `_tableInfoMap` is also in read only mode at the time. I did not see a chance that same segment's loading/downloading could be started twice.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980098956
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java:
##########
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class PredownloadZKClient implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _running;
+
+ public PredownloadZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _running = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _running = true;
+ }
+
+ @Override
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _running = false;
+ }
+ }
+
+ public boolean isStarted() {
+ return _running;
+ }
+
+ public HelixDataAccessor getDataAccessor() {
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
Review Comment:
sounds good, thanks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1905904960
##########
pinot-tools/src/main/java/org/apache/pinot/tools/predownload/PredownloadException.java:
##########
@@ -0,0 +1,29 @@
+/**
+ * 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.pinot.tools.predownload;
Review Comment:
Thanks for the suggestions. Do you have any recommended destinations? How about putting directly into pinot-server pkg?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980000557
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java:
##########
@@ -0,0 +1,63 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+
+
+public class PredownloadMetrics {
+ private static final String INSTANCE_ID_TAG = "instance_id";
+ private static final String SEGMENT_ID_TAG = "segment_id";
+ private static final long BYTES_TO_MB = 1024 * 1024;
+
+ private final ServerMetrics _serverMetrics;
+
+ public PredownloadMetrics() {
+ _serverMetrics = ServerMetrics.get();
+ }
+
+ public void segmentDownloaded(boolean succeed, String segmentName, long segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_COUNT, 1);
+ // Report download speed in MB/s, avoid divide by 0
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SEGMENT_DOWNLOAD_SPEED,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
Review Comment:
usually not, as it is using "bytes" per second as unit. It should always be a big value. If it is 0, then indicating something wrong happened.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980004355
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java:
##########
@@ -0,0 +1,63 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+
+
+public class PredownloadMetrics {
+ private static final String INSTANCE_ID_TAG = "instance_id";
+ private static final String SEGMENT_ID_TAG = "segment_id";
+ private static final long BYTES_TO_MB = 1024 * 1024;
+
+ private final ServerMetrics _serverMetrics;
+
+ public PredownloadMetrics() {
+ _serverMetrics = ServerMetrics.get();
+ }
+
+ public void segmentDownloaded(boolean succeed, String segmentName, long segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_COUNT, 1);
+ // Report download speed in MB/s, avoid divide by 0
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SEGMENT_DOWNLOAD_SPEED,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
+ } else {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_FAILURE_COUNT, 1);
+ }
+ }
+
+ public void preDownloadSucceed(long totalSegmentSizeBytes, long totalDownloadTimeMs) {
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.PREDOWNLOAD_SPEED,
+ (totalSegmentSizeBytes / BYTES_TO_MB) / (totalDownloadTimeMs / 1000 + 1));
Review Comment:
As above, it should be a big value, e.g. 200 * 10^6 / 60 * 1 (normal speed could be 1 minute to download 200MB segment). The difference between float and int is too small
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980041610
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
+ @Nullable
+ private static PredownloadMetrics _predownloadMetrics;
+
+ private PredownloadStatusRecorder() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void registerMetrics(PredownloadMetrics predownloadMetrics) {
+ PredownloadStatusRecorder._predownloadMetrics = predownloadMetrics;
+ }
+
+ @VisibleForTesting
+ static void setStatusRecordFolder(String statusRecordFolder) {
+ _statusRecordFolder = statusRecordFolder;
+ }
+
+ public static void predownloadComplete(PredownloadCompletionReason reason, String clusterName, String instanceName,
+ String segmentName) {
+ LOGGER.info(reason.getMessage(clusterName, instanceName, segmentName));
+ if (_predownloadMetrics != null) {
+ _predownloadMetrics.preDownloadComplete(reason);
+ }
+ if (reason.isSucceed()) {
+ predownloadSucceeded(reason);
+ } else {
+ predownloadFailed(reason);
+ }
+ }
+
+ private static void predownloadSucceeded(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpSuccessfulStatus(statusFolder);
+ ExitHelper.exit(0);
+ }
+
+ private static void predownloadFailed(PredownloadCompletionReason reason) {
+ if (reason.isRetriable()) {
+ predownloadRetriableFailed(reason);
+ } else {
+ predownloadNonRetriableFailed(reason);
+ }
+ }
+
+ private static void predownloadRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ checkAndDumpRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(1);
+ }
+
+ private static void predownloadNonRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpNonRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(2);
+ }
+
+ private static File prepareStatusFolder() {
+ File statusFolder = new File(_statusRecordFolder);
+ try {
+ if (statusFolder.isFile()) {
+ FileUtils.forceDelete(statusFolder);
+ }
+ FileUtils.forceMkdir(statusFolder);
+ } catch (IOException e) {
+ LOGGER.error("Failed to create status folder", e);
Review Comment:
Right now, the followup usages of the `prepareStatusFolder` also catch `IOException`. So there's no major difference between "returning null and pre-check in followup func" vs "Blindly use it and catch IOException in followup func"
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980055105
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadTableInfo.java:
##########
@@ -0,0 +1,109 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadTableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadTableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public PredownloadTableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param predownloadSegmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
Review Comment:
Good catch, removed.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978448108
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java:
##########
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class PredownloadZKClient implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _running;
+
+ public PredownloadZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _running = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _running = true;
+ }
+
+ @Override
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _running = false;
+ }
+ }
+
+ public boolean isStarted() {
+ return _running;
+ }
+
+ public HelixDataAccessor getDataAccessor() {
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
Review Comment:
if (_helixDataAccessor == null) {
_helixDataAccessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
}
return _helixDataAccessor;
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #14686:
URL: https://github.com/apache/pinot/pull/14686#issuecomment-2552600594
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/14686?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `90.90909%` with `1 line` in your changes missing coverage. Please review.
> Project coverage is 56.17%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`f23ae1a`)](https://app.codecov.io/gh/apache/pinot/commit/f23ae1a3c64f328ecd4fd94e80a4f375879228ce?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1489 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/14686?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...t/local/segment/store/SegmentLocalFSDirectory.java](https://app.codecov.io/gh/apache/pinot/pull/14686?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Fsegment%2Fstore%2FSegmentLocalFSDirectory.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9zZWdtZW50L3N0b3JlL1NlZ21lbnRMb2NhbEZTRGlyZWN0b3J5LmphdmE=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/14686?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (f23ae1a). Click for more details.
>
> <details><summary>HEAD has 48 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (f23ae1a) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|2|
>|java-21|7|2|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|1|
>|unittests|5|2|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #14686 +/- ##
============================================
- Coverage 61.75% 56.17% -5.58%
- Complexity 207 706 +499
============================================
Files 2436 2122 -314
Lines 133233 112299 -20934
Branches 20636 17846 -2790
============================================
- Hits 82274 63085 -19189
+ Misses 44911 44224 -687
+ Partials 6048 4990 -1058
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.17% <90.90%> (-5.45%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.15% <90.90%> (-5.59%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.15% <90.90%> (+28.42%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.17% <90.90%> (-5.58%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.17% <90.90%> (-5.58%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.17% <90.90%> (+9.28%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/14686/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/14686?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980056555
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadTableInfo.java:
##########
@@ -0,0 +1,109 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadTableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadTableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public PredownloadTableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param predownloadSegmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ SegmentDirectory segmentDirectory = null;
+ try {
+ segmentDirectory = getSegmentDirectory(predownloadSegmentInfo, instanceDataManagerConfig);
+ predownloadSegmentInfo.updateSegmentInfoFromLocal(segmentDirectory);
+
+ String segmentName = predownloadSegmentInfo.getSegmentName();
+ // If the segment doesn't exist on server or its CRC has changed, then we
+ // need to fall back to download the segment from deep store to load it.
+ if (!predownloadSegmentInfo.hasSameCRC()) {
+ if (predownloadSegmentInfo.getLocalCrc() == null) {
+ LOGGER.info("Segment: {} of table: {} does not exist", segmentName, _tableNameWithType);
+ } else {
+ LOGGER.info("Segment: {} of table: {} has crc change from: {} to: {}", segmentName, _tableNameWithType,
+ predownloadSegmentInfo.getLocalCrc(), predownloadSegmentInfo.getCrc());
+ }
+ return false;
+ }
+ LOGGER.info("Skip downloading segment: {} of table: {} as it already exists", segmentName, _tableNameWithType);
+ return true;
+ } finally {
+ closeSegmentDirectoryQuietly(segmentDirectory);
+ }
+ }
+
+ @Nullable
+ private SegmentDirectory getSegmentDirectory(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ String dataDir = instanceDataManagerConfig.getInstanceDataDir() + File.separator + _tableConfig.getTableName();
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(instanceDataManagerConfig, _tableConfig, _schema);
+ indexLoadingConfig.setSegmentTier(predownloadSegmentInfo.getTier());
+ indexLoadingConfig.setTableDataDir(dataDir);
+ return predownloadSegmentInfo.initSegmentDirectory(indexLoadingConfig, this);
Review Comment:
That function already `catch (Exception e)` so it would not return any exceptions.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934441447
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
Review Comment:
There are 3 types of potential bottlenecks: network, disk and CPU.
In my test environment, network is usually not unless all servers on the same rack are all doing intense IO.
CPU is used as there are many decompression jobs. But as long as each core is always assigned to decompress segments, it should be optimal.
Disk is usually the bottleneck as both downloading and decompression use disks a lot.
With num of executors > numProcessors *2, speed does not change much in my tests. *3 is good in most of times, a too high value would be even slower.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934470142
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
Review Comment:
Added here. But we are not relying on the "resource" way to close it here. We registered it in the JVM shutdown hook to call the close.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1902191664
##########
pinot-tools/src/main/java/org/apache/pinot/tools/predownload/PredownloadException.java:
##########
@@ -0,0 +1,29 @@
+/**
+ * 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.pinot.tools.predownload;
Review Comment:
If the goal is to use this predownload feature in large scale deployments then I don't think putting this in pinot-tools is the right approach. Ideally such deployments should not have a dependency on pinot-tools. pinot-tools mainly contains administrative and utility functions, and brings in a ton of dependencies.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978430512
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
Review Comment:
shall we make it configurable thread pool size?
then you can add the finding in a comment and use 3 as default.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978438743
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
+ @Nullable
+ private static PredownloadMetrics _predownloadMetrics;
+
+ private PredownloadStatusRecorder() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void registerMetrics(PredownloadMetrics predownloadMetrics) {
+ PredownloadStatusRecorder._predownloadMetrics = predownloadMetrics;
+ }
+
+ @VisibleForTesting
+ static void setStatusRecordFolder(String statusRecordFolder) {
+ _statusRecordFolder = statusRecordFolder;
+ }
+
+ public static void predownloadComplete(PredownloadCompletionReason reason, String clusterName, String instanceName,
+ String segmentName) {
+ LOGGER.info(reason.getMessage(clusterName, instanceName, segmentName));
+ if (_predownloadMetrics != null) {
+ _predownloadMetrics.preDownloadComplete(reason);
+ }
+ if (reason.isSucceed()) {
+ predownloadSucceeded(reason);
+ } else {
+ predownloadFailed(reason);
+ }
+ }
+
+ private static void predownloadSucceeded(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpSuccessfulStatus(statusFolder);
+ ExitHelper.exit(0);
+ }
+
+ private static void predownloadFailed(PredownloadCompletionReason reason) {
+ if (reason.isRetriable()) {
+ predownloadRetriableFailed(reason);
+ } else {
+ predownloadNonRetriableFailed(reason);
+ }
+ }
+
+ private static void predownloadRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ checkAndDumpRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(1);
+ }
+
+ private static void predownloadNonRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpNonRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(2);
+ }
+
+ private static File prepareStatusFolder() {
+ File statusFolder = new File(_statusRecordFolder);
+ try {
+ if (statusFolder.isFile()) {
+ FileUtils.forceDelete(statusFolder);
+ }
+ FileUtils.forceMkdir(statusFolder);
+ } catch (IOException e) {
+ LOGGER.error("Failed to create status folder", e);
Review Comment:
suggest return null if delete or mk directory failed, we will get unusable statusfolder.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978429002
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java:
##########
@@ -0,0 +1,63 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+
+
+public class PredownloadMetrics {
+ private static final String INSTANCE_ID_TAG = "instance_id";
+ private static final String SEGMENT_ID_TAG = "segment_id";
+ private static final long BYTES_TO_MB = 1024 * 1024;
+
+ private final ServerMetrics _serverMetrics;
+
+ public PredownloadMetrics() {
+ _serverMetrics = ServerMetrics.get();
+ }
+
+ public void segmentDownloaded(boolean succeed, String segmentName, long segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_COUNT, 1);
+ // Report download speed in MB/s, avoid divide by 0
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SEGMENT_DOWNLOAD_SPEED,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
+ } else {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_FAILURE_COUNT, 1);
+ }
+ }
+
+ public void preDownloadSucceed(long totalSegmentSizeBytes, long totalDownloadTimeMs) {
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.PREDOWNLOAD_SPEED,
+ (totalSegmentSizeBytes / BYTES_TO_MB) / (totalDownloadTimeMs / 1000 + 1));
+ _serverMetrics.addTimedValue(ServerTimer.PREDOWNLOAD_TIME, totalDownloadTimeMs, TimeUnit.MILLISECONDS);
+ }
+
+ public void preDownloadComplete(PredownloadCompletionReason reason) {
+ if (reason.isSucceed()) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SUCCEED, 1);
+ } else {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_FAILED, 1);
+ }
Review Comment:
_serverMetrics.addMeteredGlobalValue(
reason.isSucceed() ? ServerMeter.PREDOWNLOAD_SUCCEED : ServerMeter.PREDOWNLOAD_FAILED, 1);
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980125416
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java:
##########
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class PredownloadZKClient implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _running;
+
+ public PredownloadZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _running = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _running = true;
+ }
+
+ @Override
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _running = false;
+ }
+ }
+
+ public boolean isStarted() {
+ return _running;
+ }
+
+ public HelixDataAccessor getDataAccessor() {
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
+ return accessor;
+ }
+
+ public InstanceConfig getInstanceConfig(HelixDataAccessor accessor) {
+ String instanceConfigPath = PropertyPathBuilder.instanceConfig(_clusterName, _instanceName);
+ if (!_zkClient.exists(instanceConfigPath)) {
+ throw new HelixException("instance " + _instanceName + " does not exist in cluster " + _clusterName);
+ }
+ org.apache.helix.PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ return accessor.getProperty(keyBuilder.instanceConfig(_instanceName));
+ }
+
+ /**
+ * Get the segments of the instance from ZK. Only fetch segments that are in ONLINE state from
+ * instance current state.
+ *
+ * @return List of segment info with table name and segment name.
+ */
+ public List<PredownloadSegmentInfo> getSegmentsOfInstance(HelixDataAccessor accessor) {
+ List<PredownloadSegmentInfo> predownloadSegmentInfos = new ArrayList<>();
+ org.apache.helix.PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ LiveInstance liveInstance = accessor.getProperty(keyBuilder.liveInstance(_instanceName));
+ if (liveInstance == null) {
+ String instanceConfigPath = PropertyPathBuilder.instanceConfig(_clusterName, _instanceName);
+ PredownloadCompletionReason reason =
+ _zkClient.exists(instanceConfigPath) ? PredownloadCompletionReason.INSTANCE_NOT_ALIVE
+ : PredownloadCompletionReason.INSTANCE_NON_EXISTENT;
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceName, "");
+ } else {
+ String sessionId = liveInstance.getEphemeralOwner();
+ List<CurrentState> instanceCurrentStates =
+ accessor.getChildValues(keyBuilder.currentStates(_instanceName, sessionId), true);
+ if (instanceCurrentStates == null || instanceCurrentStates.isEmpty()) {
+ return predownloadSegmentInfos;
+ }
+ for (CurrentState currentState : instanceCurrentStates) {
+ String tableNameWithType = currentState.getResourceName();
+ Map<String, String> partitionStateMap = currentState.getPartitionStateMap();
+ for (Map.Entry<String, String> entry : partitionStateMap.entrySet()) {
+ String segmentName = entry.getKey();
+ String state = entry.getValue();
+ if (state.equals(CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE)) {
+ predownloadSegmentInfos.add(new PredownloadSegmentInfo(tableNameWithType, segmentName));
+ LOGGER.info("Added segment {} to the predownload list", segmentName);
+ } else {
+ LOGGER.info("Not add segment {} to the predownload list because it is in state {}", segmentName, state);
+ }
+ }
+ }
+ }
+ return predownloadSegmentInfos;
+ }
+
+ /**
+ * Update the segment deepstore metadata from ZK.
+ *
+ * @param predownloadSegmentInfoList List of segment info with table name and segment name.
+ * @param tableInfoMap Map of table name to table info to be filled with table config.
+ */
+ public void updateSegmentMetadata(List<PredownloadSegmentInfo> predownloadSegmentInfoList,
+ Map<String, PredownloadTableInfo> tableInfoMap,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ // fallback path comes from ZKHelixManager.class getHelixPropertyStore method
+ ZkHelixPropertyStore<ZNRecord> propertyStore = new AutoFallbackPropertyStore<>(new ZkBaseDataAccessor<>(_zkClient),
+ PropertyPathBuilder.propertyStore(_clusterName), String.format("/%s/%s", _clusterName, "HELIX_PROPERTYSTORE"));
Review Comment:
It is the path from helix https://github.com/search?q=repo%3Aapache%2Fhelix%20HELIX_PROPERTYSTORE&type=code
It is not published as parameter there, so has to make hard coded here.
And it is a fallback path that is usually not used often.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978440695
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
+ @Nullable
+ private static PredownloadMetrics _predownloadMetrics;
+
+ private PredownloadStatusRecorder() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void registerMetrics(PredownloadMetrics predownloadMetrics) {
+ PredownloadStatusRecorder._predownloadMetrics = predownloadMetrics;
+ }
+
+ @VisibleForTesting
+ static void setStatusRecordFolder(String statusRecordFolder) {
+ _statusRecordFolder = statusRecordFolder;
+ }
+
+ public static void predownloadComplete(PredownloadCompletionReason reason, String clusterName, String instanceName,
+ String segmentName) {
+ LOGGER.info(reason.getMessage(clusterName, instanceName, segmentName));
+ if (_predownloadMetrics != null) {
+ _predownloadMetrics.preDownloadComplete(reason);
+ }
+ if (reason.isSucceed()) {
+ predownloadSucceeded(reason);
+ } else {
+ predownloadFailed(reason);
+ }
+ }
+
+ private static void predownloadSucceeded(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpSuccessfulStatus(statusFolder);
+ ExitHelper.exit(0);
+ }
+
+ private static void predownloadFailed(PredownloadCompletionReason reason) {
+ if (reason.isRetriable()) {
+ predownloadRetriableFailed(reason);
+ } else {
+ predownloadNonRetriableFailed(reason);
+ }
+ }
+
+ private static void predownloadRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ checkAndDumpRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(1);
+ }
+
+ private static void predownloadNonRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpNonRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(2);
+ }
+
+ private static File prepareStatusFolder() {
+ File statusFolder = new File(_statusRecordFolder);
+ try {
+ if (statusFolder.isFile()) {
+ FileUtils.forceDelete(statusFolder);
+ }
+ FileUtils.forceMkdir(statusFolder);
+ } catch (IOException e) {
+ LOGGER.error("Failed to create status folder", e);
+ }
+ return statusFolder;
+ }
+
+ private static void dumpSuccessfulStatus(File statusFolder) {
+ try {
+ // clean all previous status first and then create a new success status file.
+ FileUtils.cleanDirectory(statusFolder);
+ File successFile = new File(statusFolder, String.format(SUCCESS_STATUS, System.currentTimeMillis()));
+ successFile.createNewFile();
Review Comment:
can file already exists? shall we check if successFile.createNewFile() == false and throw exception?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978439175
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
Review Comment:
shall we make this configurable in the next iteration?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934479163
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
Review Comment:
done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934453166
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java:
##########
@@ -169,7 +169,13 @@ public enum ServerMeter implements AbstractMetrics.Meter {
* That means that if a stage has 10 workers and all of them reach the limit, this will be increased by 1.
* But if a single query has 2 different window operators and each one reaches the limit, this will be increased by 2.
*/
- WINDOW_TIMES_MAX_ROWS_REACHED("times", true);
+ WINDOW_TIMES_MAX_ROWS_REACHED("times", true),
+
+ // predownload metrics
+ SEGMENT_DOWNLOAD_COUNT("predownloadSegmentCount", true),
Review Comment:
done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934497416
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
Review Comment:
Not in the same PR. There are too many ways to start Pinot server by different users. This could be an open discussion and resolved in follow up PR if there's some conclusions reached.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu merged PR #14686:
URL: https://github.com/apache/pinot/pull/14686
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978436532
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,430 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ // TODO: make download timeout configurable
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ private PredownloadZKClient _predownloadZkClient;
+ private List<PredownloadSegmentInfo> _predownloadSegmentInfoList;
+ private Map<String, PredownloadTableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompletionReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_predownloadZkClient != null) {
+ _predownloadZkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _predownloadZkClient = new PredownloadZKClient(_zkAddress, _clusterName, _instanceId);
+ _predownloadZkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ PredownloadStatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _predownloadSegmentInfoList = _predownloadZkClient.getSegmentsOfInstance(_predownloadZkClient.getDataAccessor());
+ if (_predownloadSegmentInfoList.isEmpty()) {
+ PredownloadCompletionReason reason = PredownloadCompletionReason.NO_SEGMENT_TO_PREDOWNLOAD;
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, "");
+ }
+ _tableInfoMap = new HashMap<>();
+ _predownloadZkClient.updateSegmentMetadata(_predownloadSegmentInfoList, _tableInfoMap,
+ _instanceDataManagerConfig);
+ }
+
+ @VisibleForTesting
+ void loadSegmentsFromLocal() {
+ LOGGER.info("Loading segments from local to reduce number of segments to download");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (PredownloadSegmentInfo predownloadSegmentInfo : _predownloadSegmentInfoList) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ boolean loadSegmentSuccess = false;
+ try {
+ PredownloadTableInfo predownloadTableInfo = _tableInfoMap.get(predownloadSegmentInfo.getTableNameWithType());
+ if (predownloadTableInfo != null) {
+ loadSegmentSuccess =
+ predownloadTableInfo.loadSegmentFromLocal(predownloadSegmentInfo, _instanceDataManagerConfig);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to load from local for segment: {} of table: {} with issue ",
+ predownloadSegmentInfo.getSegmentName(), predownloadSegmentInfo.getTableNameWithType(), e);
+ }
+ if (!loadSegmentSuccess && predownloadSegmentInfo.canBeDownloaded()) {
+ _failedSegments.add(predownloadSegmentInfo.getSegmentName());
+ }
+ }, _executor);
+
+ futures.add(future);
+ }
+
+ // Wait for all CompletableFuture tasks to complete or timeout
+ CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+
+ try {
+ // Wait indefinitely for all tasks to complete
+ allOf.get(LOAD_SEGMENTS_TIMEOUT_MIN, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ LOGGER.error("Task interrupted", e);
+ Thread.currentThread().interrupt(); // Preserve interrupted status
+ } catch (ExecutionException e) {
+ LOGGER.error("Task encountered an exception", e.getCause());
+ } catch (TimeoutException e) {
+ LOGGER.error("Task timed out", e);
+ }
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info("Load segments from local took {} sec", timeTaken / 1000);
+ }
+
+ @VisibleForTesting
+ void initializeSegmentFetcher() {
+ LOGGER.info("Initializing segment fetchers");
+ // Initialize the components to download segments from deep store
+ PinotConfiguration segmentFetcherFactoryConfig =
+ _pinotConfig.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY);
+ PinotConfiguration pinotFSConfig = _pinotConfig.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_PINOT_FS_FACTORY);
+ PinotConfiguration pinotCrypterConfig =
+ _pinotConfig.subset(CommonConstants.Server.PREFIX_OF_CONFIG_OF_PINOT_CRYPTER);
+ try {
+ SegmentFetcherFactory.init(segmentFetcherFactoryConfig);
+ PinotFSFactory.init(pinotFSConfig);
+ PinotCrypterFactory.init(pinotCrypterConfig);
+ } catch (Exception e) {
+ LOGGER.error("Failed to initialize segment fetcher factory: {}", e);
+ PredownloadStatusRecorder.predownloadComplete(PredownloadCompletionReason.CANNOT_CONNECT_TO_DEEPSTORE,
+ _clusterName,
+ _instanceId, "");
+ }
+ }
+
+ public PredownloadCompletionReason downloadSegments() {
+ LOGGER.info("Downloading segments from deep store");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (PredownloadSegmentInfo predownloadSegmentInfo : _predownloadSegmentInfoList) {
+ if (predownloadSegmentInfo.isDownloaded()) {
+ _numOfSkippedSegments++;
+ continue;
+ } else if (!predownloadSegmentInfo.canBeDownloaded()) {
+ _numOfUnableToDownloadSegments++;
+ continue;
+ } else {
+ _numOfDownloadSegments++;
+ }
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ try {
+ downloadSegment(predownloadSegmentInfo);
+ } catch (Exception e) {
+ LOGGER.error("Failed to download segment: {} of table: {} with issue ",
+ predownloadSegmentInfo.getSegmentName(),
+ predownloadSegmentInfo.getTableNameWithType(), e);
+ }
+ }, _executor);
+
+ // TODO: add future.orTimeout() to handle per segment downloading timeout
+ futures.add(future);
+ }
+
+ // Wait for all CompletableFuture tasks to complete or timeout
+ CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+
+ try {
+ // Wait indefinitely for all tasks to complete
+ allOf.get(DOWNLOAD_SEGMENTS_TIMEOUT_MIN, TimeUnit.MINUTES);
+ } catch (InterruptedException e) {
+ LOGGER.error("Task interrupted", e);
+ Thread.currentThread().interrupt(); // Preserve interrupted status
+ } catch (ExecutionException e) {
+ LOGGER.error("Task encountered an exception", e.getCause());
+ } catch (TimeoutException e) {
+ LOGGER.error("Task timed out", e);
+ }
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info("Download segments from deep store took {} sec", timeTaken / 1000);
+ return _failedSegments.isEmpty() ? PredownloadCompletionReason.ALL_SEGMENTS_DOWNLOADED
+ : PredownloadCompletionReason.SOME_SEGMENTS_DOWNLOAD_FAILED;
+ }
+
+ void downloadSegment(PredownloadSegmentInfo predownloadSegmentInfo)
+ throws Exception {
+ try {
+ long startTime = System.currentTimeMillis();
+ File tempRootDir = getTmpSegmentDataDir(predownloadSegmentInfo);
+ if (_instanceDataManagerConfig.isStreamSegmentDownloadUntar()
+ && predownloadSegmentInfo.getCrypterName() == null) {
+ try {
+ // TODO: increase rate limit here
+ File untaredSegDir = downloadAndStreamUntarWithRateLimit(predownloadSegmentInfo, tempRootDir,
+ _instanceDataManagerConfig.getStreamSegmentDownloadUntarRateLimit());
+ moveSegment(predownloadSegmentInfo, untaredSegDir);
+ } finally {
+ FileUtils.deleteQuietly(tempRootDir);
+ }
+ } else {
+ try {
+ File tarFile = downloadAndDecrypt(predownloadSegmentInfo, tempRootDir);
+ untarAndMoveSegment(predownloadSegmentInfo, tarFile, tempRootDir);
+ } finally {
+ FileUtils.deleteQuietly(tempRootDir);
+ }
Review Comment:
can we combine these two finally?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1933245880
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
Review Comment:
This timeout should be configurable. Let's leave a todo?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompleteReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _zkClient = new ZKClient(_zkAddress, _clusterName, _instanceId);
+ _zkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ StatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _segmentInfoList = _zkClient.getSegmentsOfInstance(_zkClient.getDataAccessor());
+ if (_segmentInfoList.isEmpty()) {
+ PredownloadCompleteReason reason = PredownloadCompleteReason.NO_SEGMENT_TO_PREDOWNLOAD;
Review Comment:
nit: go for "PredownloadCompletionReason", since we use similar naming elsewhere (as in segment completion protocol).
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompleteReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _zkClient = new ZKClient(_zkAddress, _clusterName, _instanceId);
+ _zkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ StatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _segmentInfoList = _zkClient.getSegmentsOfInstance(_zkClient.getDataAccessor());
+ if (_segmentInfoList.isEmpty()) {
+ PredownloadCompleteReason reason = PredownloadCompleteReason.NO_SEGMENT_TO_PREDOWNLOAD;
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, "");
+ }
+ _tableInfoMap = new HashMap<>();
+ _zkClient.updateSegmentMetadata(_segmentInfoList, _tableInfoMap, _instanceDataManagerConfig);
+ }
+
+ @VisibleForTesting
+ void loadSegmentsFromLocal() {
+ LOGGER.info("Loading segments from local to reduce number of segments to download");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (SegmentInfo segmentInfo : _segmentInfoList) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ boolean loadSegmentSuccess = false;
+ try {
+ TableInfo tableInfo = _tableInfoMap.get(segmentInfo.getTableNameWithType());
+ if (tableInfo != null) {
+ loadSegmentSuccess = tableInfo.loadSegmentFromLocal(segmentInfo, _instanceDataManagerConfig);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to load from local for segment: {} of table: {} with issue ",
+ segmentInfo.getSegmentName(), segmentInfo.getTableNameWithType(), e);
+ }
+ if (!loadSegmentSuccess && segmentInfo.canBeDownloaded()) {
+ _failedSegments.add(segmentInfo.getSegmentName());
+ }
+ }, _executor);
+
+ futures.add(future);
+ }
+
+ // Wait for all CompletableFuture tasks to complete or timeout
+ CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+
+ try {
+ // Wait indefinitely for all tasks to complete
+ allOf.get(LOAD_SEGMENTS_TIMEOUT_MIN, TimeUnit.MINUTES);
Review Comment:
what's the behavior when there are ongoing downloads and this exits? is there a possibility to run into an inconsistent/partial state on disk?
(same question for download segments step)
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ @SuppressWarnings("NullAway.Init")
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _started;
+
+ public ZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _started = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _started = true;
+ }
+
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _started = false;
Review Comment:
semantically this is incorrect, since the client was started before. Perhaps rename it to `_running`?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/TableInfo.java:
##########
@@ -0,0 +1,105 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class TableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public TableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param segmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(SegmentInfo segmentInfo, InstanceDataManagerConfig instanceDataManagerConfig) {
+ SegmentDirectory segmentDirectory = getSegmentDirectory(segmentInfo, instanceDataManagerConfig);
Review Comment:
`SegmentDirectory` is a closeable. Do we handle it somewhere?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ @SuppressWarnings("NullAway.Init")
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _started;
+
+ public ZKClient(String zkAddress, String clusterName, String instanceName) {
Review Comment:
Let's prefix all relevant classes with `Predownload` so as to not pollute namespace. (other classes `SegmentInfo`, `TableInfo`, `StatusRecorder`, `TestUtil`)
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
Review Comment:
for discussion: the downloads should largely be IO bound, so this could be much higher?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/SegmentInfo.java:
##########
@@ -0,0 +1,178 @@
+/**
+ * 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.pinot.server.predownload;
+
+import io.netty.util.internal.StringUtil;
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.config.TierConfigUtils;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoader;
+import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderContext;
+import org.apache.pinot.segment.spi.loader.SegmentDirectoryLoaderRegistry;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class SegmentInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentInfo.class);
+ private final String _segmentName;
+ private final String _tableNameWithType;
+
+ @SuppressWarnings("NullAway.Init")
Review Comment:
these should not be required here? seems like an internal repo thing
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java:
##########
@@ -169,7 +169,13 @@ public enum ServerMeter implements AbstractMetrics.Meter {
* That means that if a stage has 10 workers and all of them reach the limit, this will be increased by 1.
* But if a single query has 2 different window operators and each one reaches the limit, this will be increased by 2.
*/
- WINDOW_TIMES_MAX_ROWS_REACHED("times", true);
+ WINDOW_TIMES_MAX_ROWS_REACHED("times", true),
+
+ // predownload metrics
+ SEGMENT_DOWNLOAD_COUNT("predownloadSegmentCount", true),
Review Comment:
let's prefix the enum with `predownload` too? Since it only tracks predownload releated metrics.
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
Review Comment:
remove `e.printStackTrace()`?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
Review Comment:
This needs to be an AutoCloseable
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/StatusRecorder.java:
##########
@@ -0,0 +1,226 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+enum PredownloadCompleteReason {
Review Comment:
Separate files for each definition in this?
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
Review Comment:
We haven't really hooked this into server starter in this PR? You are planning to do a follow-up?
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java:
##########
@@ -79,7 +79,9 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
REALTIME_INGESTION_OFFSET_LAG("offsetLag", false),
REALTIME_INGESTION_UPSTREAM_OFFSET("upstreamOffset", false),
REALTIME_INGESTION_CONSUMING_OFFSET("consumingOffset", false),
- REALTIME_CONSUMER_DIR_USAGE("bytes", true);
+ REALTIME_CONSUMER_DIR_USAGE("bytes", true),
+ SEGMENT_DOWNLOAD_SPEED("bytes", true),
Review Comment:
the speed gauge will get overwritten several times leading to inaccurate obs.
better approach would be to just to track the amount of data predownloaded, and infer the rate later
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980029825
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
Review Comment:
good idea, add TODO for now.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934495974
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/TableInfo.java:
##########
@@ -0,0 +1,105 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class TableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public TableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param segmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(SegmentInfo segmentInfo, InstanceDataManagerConfig instanceDataManagerConfig) {
+ SegmentDirectory segmentDirectory = getSegmentDirectory(segmentInfo, instanceDataManagerConfig);
Review Comment:
closeSegmentDirectoryQuietly was called. I can move it to finally
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934451713
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/ZKClient.java:
##########
@@ -0,0 +1,185 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class ZKClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ @SuppressWarnings("NullAway.Init")
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _started;
+
+ public ZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _started = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _started = true;
+ }
+
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _started = false;
Review Comment:
done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980082747
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java:
##########
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class PredownloadZKClient implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _running;
+
+ public PredownloadZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _running = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _running = true;
+ }
+
+ @Override
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _running = false;
+ }
+ }
+
+ public boolean isStarted() {
+ return _running;
+ }
+
+ public HelixDataAccessor getDataAccessor() {
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
Review Comment:
Only `RealmAwareZkClient _zkClient` is doing the real connection. Both `ZKHelixDataAccessor` and `ZkBaseDataAccessor` are simple wrappers. We've already cached `RealmAwareZkClient`. I am ok with caching HelixDataAccessor or not. But should not be much difference.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1980047350
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadStatusRecorder.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// Helper class to test the System.exit() in UT
+class ExitHelper {
+ private static ExitAction _exitAction = new DefaultExitAction();
+
+ private ExitHelper() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void exit(int status) {
+ _exitAction.exit(status);
+ }
+
+ public static void setExitAction(ExitAction action) {
+ _exitAction = action;
+ }
+
+ public static void resetExitAction() {
+ _exitAction = new DefaultExitAction();
+ }
+
+ public interface ExitAction {
+ void exit(int status);
+ }
+
+ // Default implementation of ExitAction
+ private static class DefaultExitAction implements ExitAction {
+ @Override
+ public void exit(int status) {
+ System.exit(status);
+ }
+ }
+}
+
+// Record the status of the pre-download to be consumed by odin-pinot-worker.
+public class PredownloadStatusRecorder {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadStatusRecorder.class);
+ private static final long STATUS_RECORD_EXPIRATION_SEC = 3600 * 6; // 6 hours
+ // TODO: make those name configurable
+ private static final String SUCCESS_STATUS = "SUCCESS_%s";
+ private static final String FAILURE_STATUS = "FAILURE_%s";
+ private static final String NON_RETRIABLE_FAILURE_STATUS = "NONRETRIABLEFAILURE_%s";
+ private static String _statusRecordFolder = "/shared/predownload/status/";
+ @Nullable
+ private static PredownloadMetrics _predownloadMetrics;
+
+ private PredownloadStatusRecorder() {
+ throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+ }
+
+ public static void registerMetrics(PredownloadMetrics predownloadMetrics) {
+ PredownloadStatusRecorder._predownloadMetrics = predownloadMetrics;
+ }
+
+ @VisibleForTesting
+ static void setStatusRecordFolder(String statusRecordFolder) {
+ _statusRecordFolder = statusRecordFolder;
+ }
+
+ public static void predownloadComplete(PredownloadCompletionReason reason, String clusterName, String instanceName,
+ String segmentName) {
+ LOGGER.info(reason.getMessage(clusterName, instanceName, segmentName));
+ if (_predownloadMetrics != null) {
+ _predownloadMetrics.preDownloadComplete(reason);
+ }
+ if (reason.isSucceed()) {
+ predownloadSucceeded(reason);
+ } else {
+ predownloadFailed(reason);
+ }
+ }
+
+ private static void predownloadSucceeded(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpSuccessfulStatus(statusFolder);
+ ExitHelper.exit(0);
+ }
+
+ private static void predownloadFailed(PredownloadCompletionReason reason) {
+ if (reason.isRetriable()) {
+ predownloadRetriableFailed(reason);
+ } else {
+ predownloadNonRetriableFailed(reason);
+ }
+ }
+
+ private static void predownloadRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ checkAndDumpRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(1);
+ }
+
+ private static void predownloadNonRetriableFailed(PredownloadCompletionReason reason) {
+ File statusFolder = prepareStatusFolder();
+ dumpNonRetriableFailureStatus(statusFolder);
+ ExitHelper.exit(2);
+ }
+
+ private static File prepareStatusFolder() {
+ File statusFolder = new File(_statusRecordFolder);
+ try {
+ if (statusFolder.isFile()) {
+ FileUtils.forceDelete(statusFolder);
+ }
+ FileUtils.forceMkdir(statusFolder);
+ } catch (IOException e) {
+ LOGGER.error("Failed to create status folder", e);
+ }
+ return statusFolder;
+ }
+
+ private static void dumpSuccessfulStatus(File statusFolder) {
+ try {
+ // clean all previous status first and then create a new success status file.
+ FileUtils.cleanDirectory(statusFolder);
+ File successFile = new File(statusFolder, String.format(SUCCESS_STATUS, System.currentTimeMillis()));
+ successFile.createNewFile();
Review Comment:
If file already exists, then user does not need to run this predownload job again.
And even if they run it, it should return very fast with same "successful status". The current behavior is closer to be idempotent.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978431093
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,430 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ // TODO: make download timeout configurable
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ private PredownloadZKClient _predownloadZkClient;
+ private List<PredownloadSegmentInfo> _predownloadSegmentInfoList;
+ private Map<String, PredownloadTableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompletionReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
Review Comment:
If _totalDownloadedSizeBytes is small, this can result in 0 MB/s. shall we use float points?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978444329
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadTableInfo.java:
##########
@@ -0,0 +1,109 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadTableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadTableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public PredownloadTableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param predownloadSegmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
Review Comment:
do we need `InstanceDataManagerConfig instanceDataManagerConfig` in method parameter?
can we use _instanceDataManagerConfig directly?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978426979
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java:
##########
@@ -0,0 +1,63 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+
+
+public class PredownloadMetrics {
+ private static final String INSTANCE_ID_TAG = "instance_id";
+ private static final String SEGMENT_ID_TAG = "segment_id";
+ private static final long BYTES_TO_MB = 1024 * 1024;
+
+ private final ServerMetrics _serverMetrics;
+
+ public PredownloadMetrics() {
+ _serverMetrics = ServerMetrics.get();
+ }
+
+ public void segmentDownloaded(boolean succeed, String segmentName, long segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_COUNT, 1);
+ // Report download speed in MB/s, avoid divide by 0
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SEGMENT_DOWNLOAD_SPEED,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
+ } else {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_FAILURE_COUNT, 1);
+ }
+ }
+
+ public void preDownloadSucceed(long totalSegmentSizeBytes, long totalDownloadTimeMs) {
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.PREDOWNLOAD_SPEED,
+ (totalSegmentSizeBytes / BYTES_TO_MB) / (totalDownloadTimeMs / 1000 + 1));
Review Comment:
same here.. should we cast to double?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978426661
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java:
##########
@@ -0,0 +1,63 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+
+
+public class PredownloadMetrics {
+ private static final String INSTANCE_ID_TAG = "instance_id";
+ private static final String SEGMENT_ID_TAG = "segment_id";
+ private static final long BYTES_TO_MB = 1024 * 1024;
+
+ private final ServerMetrics _serverMetrics;
+
+ public PredownloadMetrics() {
+ _serverMetrics = ServerMetrics.get();
+ }
+
+ public void segmentDownloaded(boolean succeed, String segmentName, long segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+ _serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_SEGMENT_DOWNLOAD_COUNT, 1);
+ // Report download speed in MB/s, avoid divide by 0
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SEGMENT_DOWNLOAD_SPEED,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
Review Comment:
can this lead to integer division issue? e.g. return 0 for small segment size.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934444701
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
Review Comment:
sure, done
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,423 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ @SuppressWarnings("NullAway.Init")
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ @SuppressWarnings("NullAway.Init")
+ private ZKClient _zkClient;
+ @SuppressWarnings("NullAway.Init")
+ private List<SegmentInfo> _segmentInfoList;
+ @SuppressWarnings("NullAway.Init")
+ private Map<String, TableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompleteReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ StatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _zkClient = new ZKClient(_zkAddress, _clusterName, _instanceId);
+ _zkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ StatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _segmentInfoList = _zkClient.getSegmentsOfInstance(_zkClient.getDataAccessor());
+ if (_segmentInfoList.isEmpty()) {
+ PredownloadCompleteReason reason = PredownloadCompleteReason.NO_SEGMENT_TO_PREDOWNLOAD;
Review Comment:
done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934448578
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/StatusRecorder.java:
##########
@@ -0,0 +1,226 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.FileFilterUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+enum PredownloadCompleteReason {
Review Comment:
done
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1934500253
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java:
##########
@@ -79,7 +79,9 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
REALTIME_INGESTION_OFFSET_LAG("offsetLag", false),
REALTIME_INGESTION_UPSTREAM_OFFSET("upstreamOffset", false),
REALTIME_INGESTION_CONSUMING_OFFSET("consumingOffset", false),
- REALTIME_CONSUMER_DIR_USAGE("bytes", true);
+ REALTIME_CONSUMER_DIR_USAGE("bytes", true),
+ SEGMENT_DOWNLOAD_SPEED("bytes", true),
Review Comment:
It is only called once in preDownloadSucceed. We only report this once per thread. So this value would not be overriden. Emitting this value here is to have better aggregation view among 100s servers.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #14686:
URL: https://github.com/apache/pinot/pull/14686#issuecomment-2682921049
> > `feature` `performance` #14592
> > Add a pre-download feature to enable "graceful" node replacement on Pinot. With this feature, during node replacement, admins would replace the old node (OD) to new node (NN) with the same instance id. Instead of bringing down the ON before starting up NN (which is required because there cannot be 2 nodes taking same helix id at the same time), admins could:
> >
> > 1. Start NN in "pre-download" mode by adding one more parameter to StartServerCommand, like:
> >
> > ```
> > PropertiesConfiguration properties = CommonsConfigurationUtils.fromPath(<config_path>);
> > PredownloadScheduler predownloadScheduler = new PredownloadScheduler(properties);
> > predownloadScheduler.start();
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > 2. Waiting for NN "pre-download" complete with one of following conditions:
> >
> > * pre-download fully succeed
> > * pre-download partially succeed but have retried enough times
> > * pre-download failed in non-retriable mode
> > * already waited for a max time period
> > 3. Bring down the ON
> > 4. Start NN in normal mode
>
> Does it mean the admin need to change the config again and restart the NN in step 4?
Node config (xxx.properties) does not need to be changed. But admin has to shut down the NN's "pre-download" mode process and restart the node in "normal" mode. Also updated the PR description with more clarity.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on PR #14686:
URL: https://github.com/apache/pinot/pull/14686#issuecomment-2682903021
> `feature` `performance` #14592
>
> Add a pre-download feature to enable "graceful" node replacement on Pinot. With this feature, during node replacement, admins would replace the old node (OD) to new node (NN) with the same instance id. Instead of bringing down the ON before starting up NN (which is required because there cannot be 2 nodes taking same helix id at the same time), admins could:
>
> 1. Start NN in "pre-download" mode by adding one more parameter to StartServerCommand, like:
>
> ```
> PropertiesConfiguration properties = CommonsConfigurationUtils.fromPath(<config_path>);
> PredownloadScheduler predownloadScheduler = new PredownloadScheduler(properties);
> predownloadScheduler.start();
> ```
>
> 2. Waiting for NN "pre-download" complete with one of following conditions:
> - pre-download fully succeed
> - pre-download partially succeed but have retried enough times
> - pre-download failed in non-retriable mode
> - already waited for a max time period
> 3. Bring down the ON
> 4. Start NN in normal mode
Does it mean the admin need to change the config again and restart the NN in step 4?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978448778
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java:
##########
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixException;
+import org.apache.helix.PropertyPathBuilder;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.store.zk.AutoFallbackPropertyStore;
+import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.api.client.RealmAwareZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
+import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class is to manage the ZK connection and operations. It will be used to fetch the segment
+ * metadata from ZK and prepare for the downloading.
+ */
+public class PredownloadZKClient implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadZKClient.class);
+ private static final long ZK_CONNECTION_TIMEOUT_MS = 30000L;
+ private final String _clusterName;
+ private final String _instanceName;
+ private final String _zkAddress;
+
+ private RealmAwareZkClient _zkClient;
+
+ private boolean _running;
+
+ public PredownloadZKClient(String zkAddress, String clusterName, String instanceName) {
+ _clusterName = clusterName;
+ _instanceName = instanceName;
+ _zkAddress = zkAddress;
+ _running = false;
+ }
+
+ public void start() {
+ RealmAwareZkClient.RealmAwareZkClientConfig config = new RealmAwareZkClient.RealmAwareZkClientConfig();
+ config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
+ config.setZkSerializer(new ZNRecordSerializer());
+ _zkClient = SharedZkClientFactory.getInstance()
+ .buildZkClient(new HelixZkClient.ZkConnectionConfig(_zkAddress), config.createHelixZkClientConfig());
+ _zkClient.waitUntilConnected(ZK_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ _running = true;
+ }
+
+ @Override
+ public void close() {
+ if (_zkClient != null) {
+ _zkClient.close();
+ _running = false;
+ }
+ }
+
+ public boolean isStarted() {
+ return _running;
+ }
+
+ public HelixDataAccessor getDataAccessor() {
+ HelixDataAccessor accessor = new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor(_zkClient));
+ return accessor;
+ }
+
+ public InstanceConfig getInstanceConfig(HelixDataAccessor accessor) {
+ String instanceConfigPath = PropertyPathBuilder.instanceConfig(_clusterName, _instanceName);
+ if (!_zkClient.exists(instanceConfigPath)) {
+ throw new HelixException("instance " + _instanceName + " does not exist in cluster " + _clusterName);
+ }
+ org.apache.helix.PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ return accessor.getProperty(keyBuilder.instanceConfig(_instanceName));
+ }
+
+ /**
+ * Get the segments of the instance from ZK. Only fetch segments that are in ONLINE state from
+ * instance current state.
+ *
+ * @return List of segment info with table name and segment name.
+ */
+ public List<PredownloadSegmentInfo> getSegmentsOfInstance(HelixDataAccessor accessor) {
+ List<PredownloadSegmentInfo> predownloadSegmentInfos = new ArrayList<>();
+ org.apache.helix.PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ LiveInstance liveInstance = accessor.getProperty(keyBuilder.liveInstance(_instanceName));
+ if (liveInstance == null) {
+ String instanceConfigPath = PropertyPathBuilder.instanceConfig(_clusterName, _instanceName);
+ PredownloadCompletionReason reason =
+ _zkClient.exists(instanceConfigPath) ? PredownloadCompletionReason.INSTANCE_NOT_ALIVE
+ : PredownloadCompletionReason.INSTANCE_NON_EXISTENT;
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceName, "");
+ } else {
+ String sessionId = liveInstance.getEphemeralOwner();
+ List<CurrentState> instanceCurrentStates =
+ accessor.getChildValues(keyBuilder.currentStates(_instanceName, sessionId), true);
+ if (instanceCurrentStates == null || instanceCurrentStates.isEmpty()) {
+ return predownloadSegmentInfos;
+ }
+ for (CurrentState currentState : instanceCurrentStates) {
+ String tableNameWithType = currentState.getResourceName();
+ Map<String, String> partitionStateMap = currentState.getPartitionStateMap();
+ for (Map.Entry<String, String> entry : partitionStateMap.entrySet()) {
+ String segmentName = entry.getKey();
+ String state = entry.getValue();
+ if (state.equals(CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE)) {
+ predownloadSegmentInfos.add(new PredownloadSegmentInfo(tableNameWithType, segmentName));
+ LOGGER.info("Added segment {} to the predownload list", segmentName);
+ } else {
+ LOGGER.info("Not add segment {} to the predownload list because it is in state {}", segmentName, state);
+ }
+ }
+ }
+ }
+ return predownloadSegmentInfos;
+ }
+
+ /**
+ * Update the segment deepstore metadata from ZK.
+ *
+ * @param predownloadSegmentInfoList List of segment info with table name and segment name.
+ * @param tableInfoMap Map of table name to table info to be filled with table config.
+ */
+ public void updateSegmentMetadata(List<PredownloadSegmentInfo> predownloadSegmentInfoList,
+ Map<String, PredownloadTableInfo> tableInfoMap,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ // fallback path comes from ZKHelixManager.class getHelixPropertyStore method
+ ZkHelixPropertyStore<ZNRecord> propertyStore = new AutoFallbackPropertyStore<>(new ZkBaseDataAccessor<>(_zkClient),
+ PropertyPathBuilder.propertyStore(_clusterName), String.format("/%s/%s", _clusterName, "HELIX_PROPERTYSTORE"));
Review Comment:
`HELIX_PROPERTYSTORE` is this a constant introduced in this PR?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978444790
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadTableInfo.java:
##########
@@ -0,0 +1,109 @@
+/**
+ * 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.pinot.server.predownload;
+
+import java.io.File;
+import javax.annotation.Nullable;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadTableInfo {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadTableInfo.class);
+ private final String _tableNameWithType;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final TableConfig _tableConfig;
+ @Nullable
+ private final Schema _schema;
+
+ public PredownloadTableInfo(String tableNameWithType, TableConfig tableConfig, @Nullable Schema schema,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ _tableNameWithType = tableNameWithType;
+ _tableConfig = tableConfig;
+ _schema = schema;
+ _instanceDataManagerConfig = instanceDataManagerConfig;
+ }
+
+ private static void closeSegmentDirectoryQuietly(@Nullable SegmentDirectory segmentDirectory) {
+ if (segmentDirectory != null) {
+ try {
+ segmentDirectory.close();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close SegmentDirectory due to error: {}", e.getMessage());
+ }
+ }
+ }
+
+ public TableConfig getTableConfig() {
+ return _tableConfig;
+ }
+
+ public InstanceDataManagerConfig getInstanceDataManagerConfig() {
+ return _instanceDataManagerConfig;
+ }
+
+ /**
+ * After loading segment metadata from ZK, try to load from local and check if we are able to skip
+ * the downloading
+ *
+ * @param predownloadSegmentInfo SegmentInfo of segment to be loaded
+ * @param instanceDataManagerConfig InstanceDataManagerConfig loaded from scheduler
+ * @return true if already presents, false if needs to be downloaded
+ */
+ public boolean loadSegmentFromLocal(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ SegmentDirectory segmentDirectory = null;
+ try {
+ segmentDirectory = getSegmentDirectory(predownloadSegmentInfo, instanceDataManagerConfig);
+ predownloadSegmentInfo.updateSegmentInfoFromLocal(segmentDirectory);
+
+ String segmentName = predownloadSegmentInfo.getSegmentName();
+ // If the segment doesn't exist on server or its CRC has changed, then we
+ // need to fall back to download the segment from deep store to load it.
+ if (!predownloadSegmentInfo.hasSameCRC()) {
+ if (predownloadSegmentInfo.getLocalCrc() == null) {
+ LOGGER.info("Segment: {} of table: {} does not exist", segmentName, _tableNameWithType);
+ } else {
+ LOGGER.info("Segment: {} of table: {} has crc change from: {} to: {}", segmentName, _tableNameWithType,
+ predownloadSegmentInfo.getLocalCrc(), predownloadSegmentInfo.getCrc());
+ }
+ return false;
+ }
+ LOGGER.info("Skip downloading segment: {} of table: {} as it already exists", segmentName, _tableNameWithType);
+ return true;
+ } finally {
+ closeSegmentDirectoryQuietly(segmentDirectory);
+ }
+ }
+
+ @Nullable
+ private SegmentDirectory getSegmentDirectory(PredownloadSegmentInfo predownloadSegmentInfo,
+ InstanceDataManagerConfig instanceDataManagerConfig) {
+ String dataDir = instanceDataManagerConfig.getInstanceDataDir() + File.separator + _tableConfig.getTableName();
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(instanceDataManagerConfig, _tableConfig, _schema);
+ indexLoadingConfig.setSegmentTier(predownloadSegmentInfo.getTier());
+ indexLoadingConfig.setTableDataDir(dataDir);
+ return predownloadSegmentInfo.initSegmentDirectory(indexLoadingConfig, this);
Review Comment:
shall we try catch `initSegmentDirectory`?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "deemoliu (via GitHub)" <gi...@apache.org>.
deemoliu commented on code in PR #14686:
URL: https://github.com/apache/pinot/pull/14686#discussion_r1978434991
##########
pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java:
##########
@@ -0,0 +1,430 @@
+/**
+ * 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.pinot.server.predownload;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
+import org.apache.pinot.server.conf.ServerConf;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.crypt.PinotCrypterFactory;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class PredownloadScheduler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PredownloadScheduler.class);
+ private static final String TMP_DIR_NAME = "tmp";
+ // Segment download dir in format of "tmp-" + segmentName + "-" + UUID.randomUUID()
+ private static final String TMP_DIR_FORMAT = "tmp-%s-%s";
+ // TODO: make download timeout configurable
+ private static final long DOWNLOAD_SEGMENTS_TIMEOUT_MIN = 60;
+ private static final long LOAD_SEGMENTS_TIMEOUT_MIN = 5;
+ private final PropertiesConfiguration _properties;
+ private final PinotConfiguration _pinotConfig;
+ private final InstanceDataManagerConfig _instanceDataManagerConfig;
+ private final String _clusterName;
+ private final String _instanceId;
+ private final String _zkAddress;
+ @VisibleForTesting
+ Executor _executor;
+ @VisibleForTesting
+ Set<String> _failedSegments;
+ private PredownloadMetrics _predownloadMetrics;
+ private int _numOfSkippedSegments;
+ private int _numOfUnableToDownloadSegments;
+ private int _numOfDownloadSegments;
+ private long _totalDownloadedSizeBytes;
+ private PredownloadZKClient _predownloadZkClient;
+ private List<PredownloadSegmentInfo> _predownloadSegmentInfoList;
+ private Map<String, PredownloadTableInfo> _tableInfoMap;
+
+ public PredownloadScheduler(PropertiesConfiguration properties)
+ throws Exception {
+ _properties = properties;
+ _clusterName = properties.getString(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
+ _zkAddress = properties.getString(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
+ _instanceId = properties.getString(CommonConstants.Server.CONFIG_OF_INSTANCE_ID);
+ _pinotConfig = new PinotConfiguration(properties);
+ _instanceDataManagerConfig =
+ new HelixInstanceDataManagerConfig(new ServerConf(_pinotConfig).getInstanceDataManagerConfig());
+ // Get the number of available processors (vCPUs)
+ int numProcessors = Runtime.getRuntime().availableProcessors();
+ _failedSegments = ConcurrentHashMap.newKeySet();
+ // TODO: tune the value
+ _executor = Executors.newFixedThreadPool(numProcessors * 3);
+ LOGGER.info("Created thread pool with num of threads: {}", numProcessors * 3);
+ _numOfSkippedSegments = 0;
+ _numOfDownloadSegments = 0;
+ }
+
+ public void start() {
+
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.info("Trying to stop predownload process!");
+ stop();
+ } catch (Exception e) {
+ LOGGER.error("error shutting down predownload process : ", e);
+ }
+ }
+ });
+
+ long startTime = System.currentTimeMillis();
+ initializeZK();
+ initializeMetricsReporter();
+ initializeSegmentFetcher();
+ getSegmentsInfo();
+ loadSegmentsFromLocal();
+ PredownloadCompletionReason reason = downloadSegments();
+ long timeTaken = System.currentTimeMillis() - startTime;
+ LOGGER.info(
+ "Predownload process took {} sec, tried to download {} segments, skipped {} segments "
+ + "and unable to download {} segments. Download size: {} MB. Download speed: {} MB/s",
+ timeTaken / 1000, _numOfDownloadSegments, _numOfSkippedSegments, _numOfUnableToDownloadSegments,
+ _totalDownloadedSizeBytes / (1024 * 1024),
+ (_totalDownloadedSizeBytes / (1024 * 1024)) / (timeTaken / 1000 + 1));
+ if (reason.isSucceed()) {
+ _predownloadMetrics.preDownloadSucceed(_totalDownloadedSizeBytes, timeTaken);
+ }
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, String.join(",", _failedSegments));
+ }
+
+ public void stop() {
+ if (_predownloadZkClient != null) {
+ _predownloadZkClient.close();
+ }
+ if (_executor != null) {
+ ((ThreadPoolExecutor) _executor).shutdownNow();
+ }
+ }
+
+ void initializeZK() {
+ LOGGER.info("Initializing ZK client with address: {} and instanceId: {}", _zkAddress, _instanceId);
+ _predownloadZkClient = new PredownloadZKClient(_zkAddress, _clusterName, _instanceId);
+ _predownloadZkClient.start();
+ }
+
+ void initializeMetricsReporter() {
+ LOGGER.info("Initializing metrics reporter");
+
+ _predownloadMetrics = new PredownloadMetrics();
+ PredownloadStatusRecorder.registerMetrics(_predownloadMetrics);
+ }
+
+ @VisibleForTesting
+ void getSegmentsInfo() {
+ LOGGER.info("Getting segments info from ZK");
+ _predownloadSegmentInfoList = _predownloadZkClient.getSegmentsOfInstance(_predownloadZkClient.getDataAccessor());
+ if (_predownloadSegmentInfoList.isEmpty()) {
+ PredownloadCompletionReason reason = PredownloadCompletionReason.NO_SEGMENT_TO_PREDOWNLOAD;
+ PredownloadStatusRecorder.predownloadComplete(reason, _clusterName, _instanceId, "");
+ }
+ _tableInfoMap = new HashMap<>();
+ _predownloadZkClient.updateSegmentMetadata(_predownloadSegmentInfoList, _tableInfoMap,
+ _instanceDataManagerConfig);
+ }
+
+ @VisibleForTesting
+ void loadSegmentsFromLocal() {
+ LOGGER.info("Loading segments from local to reduce number of segments to download");
+ long startTime = System.currentTimeMillis();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ // Submit tasks to the executor
+ for (PredownloadSegmentInfo predownloadSegmentInfo : _predownloadSegmentInfoList) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ boolean loadSegmentSuccess = false;
+ try {
+ PredownloadTableInfo predownloadTableInfo = _tableInfoMap.get(predownloadSegmentInfo.getTableNameWithType());
+ if (predownloadTableInfo != null) {
+ loadSegmentSuccess =
+ predownloadTableInfo.loadSegmentFromLocal(predownloadSegmentInfo, _instanceDataManagerConfig);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed to load from local for segment: {} of table: {} with issue ",
+ predownloadSegmentInfo.getSegmentName(), predownloadSegmentInfo.getTableNameWithType(), e);
Review Comment:
there are some repeated table lookup, can we reuse `predownloadSegmentInfo.getTableNameWithType()`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add predownload functionality to Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #14686:
URL: https://github.com/apache/pinot/pull/14686#issuecomment-2737641601
Thanks for the contribution! Can you also add documentation about how to use this feature?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org