This is an automated email from the ASF dual-hosted git repository.
shauryachats pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1f1521b8389 Add peer download fallback to PredownloadScheduler (#18641)
1f1521b8389 is described below
commit 1f1521b83890932bd05b49feb2fdbdae631c2ded
Author: Pradeep Singh Negi <[email protected]>
AuthorDate: Tue Jul 7 12:34:37 2026 +0530
Add peer download fallback to PredownloadScheduler (#18641)
* Add peer download fallback to PredownloadScheduler
When a segment download from deep storage fails in the predownload
container, fall back to downloading from peer servers if peer download
is enabled. This improves predownload reliability when deep storage is
temporarily unavailable or experiencing issues.
Changes:
- Accept peerDownloadEnabled flag in PredownloadScheduler constructor
- Extract deep store download logic into downloadFromDeepStore()
- Add downloadFromPeers() method that discovers ONLINE peer servers
via ExternalView and downloads the segment from a shuffled peer list
- Add getPeerServerURIs() to PredownloadZKClient to discover peers
without requiring HelixManager
- Add tests for peer download fallback and ZK peer discovery
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
* Read peerDownloadEnabled from properties instead of constructor parameter
The upstream StartServer now sets pinot.server.peer.download.enabled
in properties, so the constructor no longer needs the boolean parameter.
This simplifies the API and keeps configuration in one place.
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
* Address review comments: sanitize peer download scheme and fix Javadoc
1. Sanitize _peerDownloadScheme similar to BaseTableDataManager:
lowercase the value and validate it's http or https.
2. Remove "excluding the current instance" from getPeerServerURIs
Javadoc — predownload runs before server starts, so the current
instance is never in ExternalView.
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
* Add peer download and deep store metrics for predownload
- PREDOWNLOAD_DEEPSTORE_DOWNLOAD_COUNT: deep store success counter
- PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_COUNT: peer download success counter
- PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_FAILURE_COUNT: peer download failure
counter with segment name as key for per-segment tracking
- PEER_DOWNLOAD_SPEED: gauge for peer download speed (MB/s)
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
* Add tests for peer download and deep store metrics
- testDeepStoreSuccessEmitsDeepStoreMetric: verifies
deepStoreSegmentDownloaded()
is called on deep store success, peer metrics not called
- testPeerDownloadSuccessEmitsPeerMetrics: verifies
peerSegmentDownloaded(true)
and segmentDownloaded(true) both called on peer fallback success
- testPeerDownloadFailureEmitsSegmentLevelMetric: verifies
peerSegmentDownloaded(false, segmentName) called with segment name on
failure
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
* Add streaming-untar support for peer download in predownload
When streaming download-untar is enabled and the segment is not encrypted,
peer download now uses the streaming path (fetchAndStreamUntarToLocal)
instead of download-then-untar, matching the existing deep store behavior.
Framework changes:
- Add fetchUntarSegmentToLocalStreamed(segmentName, uriSupplier, dest,
rateLimit)
to SegmentFetcher interface, BaseSegmentFetcher (with retry), and
HttpSegmentFetcher
- Add corresponding SegmentFetcherFactory.fetchAndStreamUntarToLocal
overload
Tests added for HttpSegmentFetcher, SegmentFetcherFactory, and
PredownloadScheduler streaming peer fallback path.
* Address review comments: metric naming, config constant, guard refactor
- Rename PEER_DOWNLOAD_SPEED to PEER_DOWNLOAD_SPEED_MBPS with unit "mbps"
- Remove segmentName from failure metric to avoid high cardinality
- Move hardcoded "pinot.server.peer.download.enabled" to
CommonConstants.Server.CONFIG_OF_PEER_DOWNLOAD_ENABLED
- Move peer download enabled/scheme guard into downloadFromPeers
- On peer download failure, throw the original deep store exception
---------
Co-authored-by: psinghnegi <[email protected]>
Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
---
.../apache/pinot/common/metrics/ServerGauge.java | 1 +
.../apache/pinot/common/metrics/ServerMeter.java | 3 +
.../common/utils/fetcher/BaseSegmentFetcher.java | 39 ++
.../common/utils/fetcher/HttpSegmentFetcher.java | 16 +
.../pinot/common/utils/fetcher/SegmentFetcher.java | 6 +
.../utils/fetcher/SegmentFetcherFactory.java | 10 +
.../utils/fetcher/HttpSegmentFetcherTest.java | 52 +++
.../utils/fetcher/SegmentFetcherFactoryTest.java | 16 +
.../server/predownload/PredownloadMetrics.java | 14 +
.../server/predownload/PredownloadScheduler.java | 110 +++++-
.../server/predownload/PredownloadZKClient.java | 47 +++
.../server/predownload/PredownloadMetricsTest.java | 64 +++
.../predownload/PredownloadSchedulerTest.java | 430 ++++++++++++++++++++-
.../predownload/PredownloadZKClientTest.java | 72 ++++
.../apache/pinot/spi/utils/CommonConstants.java | 2 +
15 files changed, 865 insertions(+), 17 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
index ac29c9541bd..0a909ce4341 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
@@ -106,6 +106,7 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
REALTIME_CONSUMER_DIR_USAGE("bytes", true),
SEGMENT_DOWNLOAD_SPEED("bytes", true),
PREDOWNLOAD_SPEED("bytes", true),
+ PEER_DOWNLOAD_SPEED_MBPS("mbps", true),
ZK_JUTE_MAX_BUFFER("zkJuteMaxBuffer", true),
// gRPC Netty buffer metrics
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
index c5dc9855006..81ff8c3df82 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
@@ -222,6 +222,9 @@ public enum ServerMeter implements AbstractMetrics.Meter {
PREDOWNLOAD_SEGMENT_DOWNLOAD_FAILURE_COUNT("predownloadSegmentFailureCount",
true),
PREDOWNLOAD_SUCCEED("predownloadSucceed", true),
PREDOWNLOAD_FAILED("predownloadFailed", true),
+ PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_COUNT("predownloadPeerSegmentCount", true),
+
PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_FAILURE_COUNT("predownloadPeerSegmentFailureCount",
false),
+ PREDOWNLOAD_DEEPSTORE_DOWNLOAD_COUNT("predownloadDeepstoreDownloadCount",
true),
// reingestion metrics
SEGMENT_REINGESTION_FAILURE("segments", false),
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/BaseSegmentFetcher.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/BaseSegmentFetcher.java
index 5fb82388f2b..0cdcc670e00 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/BaseSegmentFetcher.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/BaseSegmentFetcher.java
@@ -22,6 +22,7 @@ import java.io.File;
import java.net.URI;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.utils.RoundRobinURIProvider;
@@ -151,4 +152,42 @@ public abstract class BaseSegmentFetcher implements
SegmentFetcher {
throws Exception {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public File fetchUntarSegmentToLocalStreamed(String segmentName,
Supplier<List<URI>> uriSupplier, File dest,
+ long maxStreamRateInByte) throws Exception {
+ AtomicReference<File> ret = new AtomicReference<>();
+ try {
+ int attempt =
+ RetryPolicies.exponentialBackoffRetryPolicy(_retryCount,
_retryWaitMs, _retryDelayScaleFactor).attempt(() -> {
+ List<URI> suppliedURIs = uriSupplier.get();
+ // Go through the list of URIs to fetch and untar the segment
until success.
+ for (URI uri : suppliedURIs) {
+ try {
+ ret.set(fetchUntarSegmentToLocalWithoutRetry(uri, dest,
maxStreamRateInByte));
+ return true;
+ } catch (Exception e) {
+ _logger.warn("Stream download-untar segment {} from peer {}
failed.", segmentName, uri, e);
+ }
+ }
+ // None of the URI works. Return false for retry.
+ return false;
+ });
+ _logger.info("Stream downloaded and untarred segment {} successfully
with {} attempts.", segmentName,
+ attempt + 1);
+ return ret.get();
+ } catch (Exception e) {
+ _logger.error("Failed to stream download-untar segment {} after
retries.", segmentName, e);
+ throw e;
+ }
+ }
+
+ /**
+ * Fetches a segment from URI location and untars it to local in a streamed
manner, without retry. Sub-class
+ * should override this to support {@link
#fetchUntarSegmentToLocalStreamed(String, Supplier, File, long)}.
+ */
+ protected File fetchUntarSegmentToLocalWithoutRetry(URI uri, File dest, long
maxStreamRateInByte)
+ throws Exception {
+ throw new UnsupportedOperationException();
+ }
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java
index 010b43eb777..dbb37455bc7 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcher.java
@@ -200,6 +200,22 @@ public class HttpSegmentFetcher extends BaseSegmentFetcher
{
return ret.get();
}
+ @Override
+ protected File fetchUntarSegmentToLocalWithoutRetry(URI uri, File dest, long
maxStreamRateInByte)
+ throws Exception {
+ String hostName = uri.getHost();
+ int port = uri.getPort();
+ // If the original download address is specified as host name, need add a
"HOST" HTTP header to the HTTP
+ // request. Otherwise, if the download address is a LB address, when the
LB be configured as "disallow direct
+ // access by IP address", downloading will fail.
+ List<Header> httpHeaders = new LinkedList<>();
+ if (!InetAddresses.isInetAddress(hostName)) {
+ httpHeaders.add(new BasicHeader(HttpHeaders.HOST, hostName + ":" +
port));
+ }
+ return _httpClient.downloadUntarFileStreamed(uri, dest, _authProvider,
httpHeaders, maxStreamRateInByte,
+ _connectionRequestTimeoutMs, _socketTimeoutMs);
+ }
+
@Override
public void fetchSegmentToLocalWithoutRetry(URI uri, File dest)
throws Exception {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcher.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcher.java
index 96961e4a089..d17086b6f84 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcher.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcher.java
@@ -58,4 +58,10 @@ public interface SegmentFetcher {
* @throws Exception when the segment fetch fails after all attempts are
exhausted or other runtime exceptions occur.
*/
void fetchSegmentToLocal(String segmentName, Supplier<List<URI>>
uriSupplier, File dest) throws Exception;
+
+ /**
+ * Fetches a segment from any uri in the given supplier's list, and untars
it to local in a streamed manner.
+ */
+ File fetchUntarSegmentToLocalStreamed(String segmentName,
Supplier<List<URI>> uriSupplier, File dest,
+ long maxStreamRateInByte) throws Exception;
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
index de93adfe38c..ee6f89cac93 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactory.java
@@ -196,4 +196,14 @@ public class SegmentFetcherFactory {
crypter.decrypt(tempDownloadedFile, dest);
}
}
+
+ /**
+ * Fetches a segment from any uri in the given supplier's list, and untars
it to local in a streamed manner.
+ */
+ public static File fetchAndStreamUntarToLocal(String segmentName, String
scheme, Supplier<List<URI>> uriSupplier,
+ File dest, long maxStreamRateInByte)
+ throws Exception {
+ return
getSegmentFetcher(scheme).fetchUntarSegmentToLocalStreamed(segmentName,
uriSupplier, dest,
+ maxStreamRateInByte);
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcherTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcherTest.java
index c83036a6332..2819ad542e2 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcherTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/HttpSegmentFetcherTest.java
@@ -19,6 +19,7 @@
package org.apache.pinot.common.utils.fetcher;
import java.io.File;
+import java.io.IOException;
import java.net.URI;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
@@ -33,6 +34,8 @@ import org.testng.annotations.Test;
import static org.apache.pinot.common.utils.fetcher.HttpSegmentFetcher.*;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -126,4 +129,53 @@ public class HttpSegmentFetcherTest {
List<URI> uris = List.of();
segmentFetcher.fetchSegmentToLocal(SEGMENT_NAME, () -> uris, SEGMENT_FILE);
}
+
+ @Test
+ public void testFetchUntarSegmentToLocalStreamedSucceedAtFirstAttempt()
+ throws Exception {
+ FileUploadDownloadClient client = mock(FileUploadDownloadClient.class);
+ when(client.downloadUntarFileStreamed(any(), any(), any(), any(),
anyLong(), anyInt(), anyInt())).thenReturn(
+ SEGMENT_FILE);
+ HttpSegmentFetcher segmentFetcher = getSegmentFetcher(client);
+ List<URI> uris = List.of(new URI("http://h1:8080";), new
URI("http://h2:8080";));
+ File result =
segmentFetcher.fetchUntarSegmentToLocalStreamed(SEGMENT_NAME, () -> uris,
SEGMENT_FILE, -1);
+ Assert.assertEquals(result, SEGMENT_FILE);
+ }
+
+ @Test(expectedExceptions = AttemptsExceededException.class)
+ public void testFetchUntarSegmentToLocalStreamedAllDownloadAttemptsFailed()
+ throws Exception {
+ FileUploadDownloadClient client = mock(FileUploadDownloadClient.class);
+ // All attempts failed
+ when(client.downloadUntarFileStreamed(any(), any(), any(), any(),
anyLong(), anyInt(), anyInt())).thenThrow(
+ new IOException("Failed to download"));
+ HttpSegmentFetcher segmentFetcher = getSegmentFetcher(client);
+ List<URI> uris = List.of(new URI("http://h1:8080";), new
URI("http://h2:8080";));
+ segmentFetcher.fetchUntarSegmentToLocalStreamed(SEGMENT_NAME, () -> uris,
SEGMENT_FILE, -1);
+ }
+
+ @Test
+ public void testFetchUntarSegmentToLocalStreamedSuccessAfterRetry()
+ throws Exception {
+ FileUploadDownloadClient client = mock(FileUploadDownloadClient.class);
+ // The first two attempts failed and the last attempt succeeded
+ when(client.downloadUntarFileStreamed(any(), any(), any(), any(),
anyLong(), anyInt(), anyInt())).thenThrow(
+ new IOException("Failed to download")).thenThrow(new
IOException("Failed to download"))
+ .thenReturn(SEGMENT_FILE);
+ HttpSegmentFetcher segmentFetcher = getSegmentFetcher(client);
+ List<URI> uris = List.of(new URI("http://h1:8080";), new
URI("http://h2:8080";));
+ File result =
segmentFetcher.fetchUntarSegmentToLocalStreamed(SEGMENT_NAME, () -> uris,
SEGMENT_FILE, -1);
+ Assert.assertEquals(result, SEGMENT_FILE);
+ }
+
+ @Test(expectedExceptions = AttemptsExceededException.class)
+ public void testFetchUntarSegmentToLocalStreamedFailureWithNoPeerServers()
+ throws Exception {
+ FileUploadDownloadClient client = mock(FileUploadDownloadClient.class);
+ when(client.downloadUntarFileStreamed(any(), any(), any(), any(),
anyLong(), anyInt(), anyInt())).thenReturn(
+ SEGMENT_FILE);
+ HttpSegmentFetcher segmentFetcher = getSegmentFetcher(client);
+ List<URI> uris = List.of();
+ segmentFetcher.fetchUntarSegmentToLocalStreamed(SEGMENT_NAME, () -> uris,
SEGMENT_FILE, -1);
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactoryTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactoryTest.java
index 8e6576f0ab7..96bed9e751c 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactoryTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/fetcher/SegmentFetcherFactoryTest.java
@@ -40,6 +40,7 @@ public class SegmentFetcherFactoryTest {
private static final String FILE_PROTOCOL = "file";
private static final String TEST_PROTOCOL = "test";
private static final String TEST_URI = "test://foo/bar";
+ private static final String SEGMENT_NAME = "testSegment";
@Test
public void testDefaultSegmentFetcherFactory() {
@@ -98,6 +99,12 @@ public class SegmentFetcherFactoryTest {
assertEquals(fakeCrypter._encryptCalled, 0);
assertEquals(fakeCrypter._originalPath, "foo/bar.enc");
assertEquals(fakeCrypter._decryptedPath, "foo/bar");
+
+ List<URI> streamedUris = List.of(new URI(TEST_URI));
+ File untaredSegDir =
SegmentFetcherFactory.fetchAndStreamUntarToLocal(SEGMENT_NAME, TEST_PROTOCOL,
+ () -> streamedUris, new File("foo/bar"), -1);
+ assertEquals(untaredSegDir, new File("fakeSegmentIndexFile"));
+ assertEquals(testFileFetcher._fetchFileToLocalCalled, 5);
}
public static class FakeSegmentFetcher implements SegmentFetcher {
@@ -135,6 +142,15 @@ public class SegmentFetcherFactoryTest {
throws Exception {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public File fetchUntarSegmentToLocalStreamed(String segmentName,
Supplier<List<URI>> uriSupplier, File dest,
+ long maxStreamRateInByte)
+ throws Exception {
+ assertEquals(uriSupplier.get(), List.of(new URI(TEST_URI)));
+ _fetchFileToLocalCalled++;
+ return new File("fakeSegmentIndexFile");
+ }
}
public static class FakePinotCrypter implements PinotCrypter {
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java
index 7fb26f79038..353f22e24ce 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadMetrics.java
@@ -47,6 +47,20 @@ public class PredownloadMetrics {
}
}
+ public void peerSegmentDownloaded(boolean succeed, String segmentName, long
segmentSizeBytes, long downloadTimeMs) {
+ if (succeed) {
+
_serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_COUNT,
1);
+
_serverMetrics.setValueOfGlobalGauge(ServerGauge.PEER_DOWNLOAD_SPEED_MBPS,
+ (segmentSizeBytes / BYTES_TO_MB) / (downloadTimeMs / 1000 + 1));
+ } else {
+
_serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_FAILURE_COUNT,
1);
+ }
+ }
+
+ public void deepStoreSegmentDownloaded() {
+
_serverMetrics.addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_DEEPSTORE_DOWNLOAD_COUNT,
1);
+ }
+
public void preDownloadSucceed(long totalSegmentSizeBytes, long
totalDownloadTimeMs) {
_serverMetrics.setValueOfGlobalGauge(ServerGauge.PREDOWNLOAD_SPEED,
(totalSegmentSizeBytes / BYTES_TO_MB) / (totalDownloadTimeMs / 1000 +
1));
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java
index 558f055f152..89fa99effd7 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadScheduler.java
@@ -19,9 +19,12 @@
package org.apache.pinot.server.predownload;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import java.io.File;
import java.io.IOException;
+import java.net.URI;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -36,6 +39,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.utils.TarCompressionUtils;
@@ -72,6 +76,8 @@ public class PredownloadScheduler {
@VisibleForTesting
Set<String> _failedSegments;
private PredownloadMetrics _predownloadMetrics;
+ private boolean _peerDownloadEnabled;
+ private String _peerDownloadScheme;
private int _numOfSkippedSegments;
private int _numOfUnableToDownloadSegments;
private int _numOfDownloadSegments;
@@ -107,6 +113,19 @@ public class PredownloadScheduler {
_failedSegments = ConcurrentHashMap.newKeySet();
_executor = Executors.newFixedThreadPool(predownloadParallelism);
LOGGER.info("Created thread pool with num of threads: {}",
predownloadParallelism);
+
+ _peerDownloadEnabled =
properties.getBoolean(CommonConstants.Server.CONFIG_OF_PEER_DOWNLOAD_ENABLED,
+ CommonConstants.Server.DEFAULT_PEER_DOWNLOAD_ENABLED);
+ _peerDownloadScheme =
_instanceDataManagerConfig.getSegmentPeerDownloadScheme();
+ if (_peerDownloadScheme != null) {
+ _peerDownloadScheme = _peerDownloadScheme.toLowerCase();
+ Preconditions.checkState(
+ CommonConstants.HTTP_PROTOCOL.equals(_peerDownloadScheme)
+ || CommonConstants.HTTPS_PROTOCOL.equals(_peerDownloadScheme),
+ "Unsupported peer download scheme: %s", _peerDownloadScheme);
+ }
+ LOGGER.info("Peer download enabled: {}, scheme: {}", _peerDownloadEnabled,
_peerDownloadScheme);
+
_numOfSkippedSegments = 0;
_numOfDownloadSegments = 0;
}
@@ -301,23 +320,24 @@ public class PredownloadScheduler {
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);
+ try {
+ downloadFromDeepStore(predownloadSegmentInfo);
+ _predownloadMetrics.deepStoreSegmentDownloaded();
+ } catch (Exception deepStoreException) {
+ if (!_peerDownloadEnabled || _peerDownloadScheme == null) {
+ throw deepStoreException;
}
- } else {
+ LOGGER.warn("Deep store download failed for segment: {} of table: {},
falling back to peer download",
+ predownloadSegmentInfo.getSegmentName(),
predownloadSegmentInfo.getTableNameWithType(), deepStoreException);
+ long peerStartTime = System.currentTimeMillis();
try {
- File tarFile = downloadAndDecrypt(predownloadSegmentInfo,
tempRootDir);
- untarAndMoveSegment(predownloadSegmentInfo, tarFile, tempRootDir);
- } finally {
- FileUtils.deleteQuietly(tempRootDir);
+ downloadFromPeers(predownloadSegmentInfo);
+ _predownloadMetrics.peerSegmentDownloaded(true,
predownloadSegmentInfo.getSegmentName(),
+ predownloadSegmentInfo.getLocalSizeBytes(),
+ System.currentTimeMillis() - peerStartTime);
+ } catch (Exception peerEx) {
+ _predownloadMetrics.peerSegmentDownloaded(false,
predownloadSegmentInfo.getSegmentName(), 0, 0);
+ throw deepStoreException;
}
}
_failedSegments.remove(predownloadSegmentInfo.getSegmentName());
@@ -336,6 +356,65 @@ public class PredownloadScheduler {
}
}
+ private void downloadFromDeepStore(PredownloadSegmentInfo
predownloadSegmentInfo)
+ throws Exception {
+ File tempRootDir = getTmpSegmentDataDir(predownloadSegmentInfo);
+ if (_instanceDataManagerConfig.isStreamSegmentDownloadUntar()
+ && predownloadSegmentInfo.getCrypterName() == null) {
+ try {
+ 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);
+ }
+ }
+ }
+
+ private void downloadFromPeers(PredownloadSegmentInfo predownloadSegmentInfo)
+ throws Exception {
+ if (!_peerDownloadEnabled || _peerDownloadScheme == null) {
+ throw new PredownloadException("Peer download is not enabled or scheme
is not configured");
+ }
+ String segmentName = predownloadSegmentInfo.getSegmentName();
+ String tableNameWithType = predownloadSegmentInfo.getTableNameWithType();
+ LOGGER.info("Downloading segment: {} of table: {} from peers using scheme:
{}", segmentName, tableNameWithType,
+ _peerDownloadScheme);
+ File tempRootDir = getTmpSegmentDataDir(predownloadSegmentInfo);
+ Supplier<List<URI>> uriSupplier = () -> {
+ List<URI> peerURIs =
+
_predownloadZkClient.getPeerServerURIs(_predownloadZkClient.getDataAccessor(),
tableNameWithType,
+ segmentName, _peerDownloadScheme);
+ Collections.shuffle(peerURIs);
+ return peerURIs;
+ };
+ try {
+ if (_instanceDataManagerConfig.isStreamSegmentDownloadUntar()
+ && predownloadSegmentInfo.getCrypterName() == null) {
+ File untaredSegDir =
SegmentFetcherFactory.fetchAndStreamUntarToLocal(segmentName,
_peerDownloadScheme,
+ uriSupplier, tempRootDir,
_instanceDataManagerConfig.getStreamSegmentDownloadUntarRateLimit());
+ LOGGER.info("Downloaded and untarred segment: {} from peers to: {}",
segmentName, untaredSegDir);
+ moveSegment(predownloadSegmentInfo, untaredSegDir);
+ } else {
+ File segmentTarFile = new File(tempRootDir, segmentName +
TarCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(segmentName,
_peerDownloadScheme, uriSupplier,
+ segmentTarFile, predownloadSegmentInfo.getCrypterName());
+ LOGGER.info("Downloaded segment: {} from peers to: {}, file length:
{}", segmentName, segmentTarFile,
+ segmentTarFile.length());
+ untarAndMoveSegment(predownloadSegmentInfo, segmentTarFile,
tempRootDir);
+ }
+ } finally {
+ FileUtils.deleteQuietly(tempRootDir);
+ }
+ }
+
private File getTmpSegmentDataDir(PredownloadSegmentInfo
predownloadSegmentInfo)
throws Exception {
PredownloadTableInfo predownloadTableInfo =
_tableInfoMap.get(predownloadSegmentInfo.getTableNameWithType());
@@ -395,7 +474,6 @@ public class PredownloadScheduler {
} catch (AttemptsExceededException e) {
LOGGER.error("Attempts exceeded when downloading segment: {} for table:
{} from: {} to: {}", segmentName,
tableNameWithType, uri, tarFile);
- // TODO: add download from peer logic
throw e;
}
}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
index c504af4a65f..ac1610c2c7d 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.server.predownload;
+import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -28,6 +29,7 @@ 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.ExternalView;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.store.zk.AutoFallbackPropertyStore;
@@ -147,6 +149,51 @@ public class PredownloadZKClient implements AutoCloseable {
return predownloadSegmentInfos;
}
+ /**
+ * Returns URIs of ONLINE peer servers hosting the given segment.
+ * Uses ExternalView to discover peers — mirrors PeerServerSegmentFinder
without requiring HelixManager.
+ */
+ public List<URI> getPeerServerURIs(HelixDataAccessor accessor, String
tableNameWithType, String segmentName,
+ String downloadScheme) {
+ org.apache.helix.PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ ExternalView externalView =
accessor.getProperty(keyBuilder.externalView(tableNameWithType));
+ if (externalView == null) {
+ LOGGER.warn("No external view for table: {} when finding peers for
segment: {}", tableNameWithType, segmentName);
+ return new ArrayList<>();
+ }
+ Map<String, String> instanceStateMap =
externalView.getStateMap(segmentName);
+ if (instanceStateMap == null) {
+ LOGGER.warn("Segment: {} not found in external view of table: {}",
segmentName, tableNameWithType);
+ return new ArrayList<>();
+ }
+ String adminPortKey = CommonConstants.HTTP_PROTOCOL.equals(downloadScheme)
+ ? CommonConstants.Helix.Instance.ADMIN_PORT_KEY
+ : CommonConstants.Helix.Instance.ADMIN_HTTPS_PORT_KEY;
+ List<URI> peerURIs = new ArrayList<>();
+ for (Map.Entry<String, String> entry : instanceStateMap.entrySet()) {
+ String instanceId = entry.getKey();
+ if
(!CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE.equals(entry.getValue()))
{
+ continue;
+ }
+ InstanceConfig instanceConfig =
accessor.getProperty(keyBuilder.instanceConfig(instanceId));
+ if (instanceConfig == null) {
+ LOGGER.warn("Failed to get instance config for peer: {}", instanceId);
+ continue;
+ }
+ String hostName = instanceConfig.getHostName();
+ int port = instanceConfig.getRecord()
+ .getIntField(adminPortKey,
CommonConstants.Server.DEFAULT_ADMIN_API_PORT);
+ try {
+ peerURIs.add(new URI(
+ String.format("%s://%s:%d/segments/%s/%s",
+ downloadScheme, hostName, port, tableNameWithType,
segmentName)));
+ } catch (Exception e) {
+ LOGGER.warn("Failed to construct peer URI for instance: {}",
instanceId, e);
+ }
+ }
+ return peerURIs;
+ }
+
/**
* Update the segment deepstore metadata from ZK.
*
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadMetricsTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadMetricsTest.java
new file mode 100644
index 00000000000..36a5ec31936
--- /dev/null
+++
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadMetricsTest.java
@@ -0,0 +1,64 @@
+/**
+ * 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 org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMeter;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.mockito.MockedStatic;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+
+public class PredownloadMetricsTest {
+
+ @Test
+ public void testPeerSegmentDownloadedFailureEmitsGlobalMeter() {
+ ServerMetrics mockServerMetrics = mock(ServerMetrics.class);
+ try (MockedStatic<ServerMetrics> serverMetricsMock =
mockStatic(ServerMetrics.class)) {
+ serverMetricsMock.when(ServerMetrics::get).thenReturn(mockServerMetrics);
+ PredownloadMetrics predownloadMetrics = new PredownloadMetrics();
+
+ predownloadMetrics.peerSegmentDownloaded(false, "testSegment", 0, 0);
+
+
verify(mockServerMetrics).addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_FAILURE_COUNT,
1);
+ verifyNoMoreInteractions(mockServerMetrics);
+ }
+ }
+
+ @Test
+ public void testPeerSegmentDownloadedSuccessEmitsGlobalMeterAndGauge() {
+ ServerMetrics mockServerMetrics = mock(ServerMetrics.class);
+ try (MockedStatic<ServerMetrics> serverMetricsMock =
mockStatic(ServerMetrics.class)) {
+ serverMetricsMock.when(ServerMetrics::get).thenReturn(mockServerMetrics);
+ PredownloadMetrics predownloadMetrics = new PredownloadMetrics();
+
+ predownloadMetrics.peerSegmentDownloaded(true, "testSegment", 1024 *
1024, 1000);
+
+
verify(mockServerMetrics).addMeteredGlobalValue(ServerMeter.PREDOWNLOAD_PEER_SEGMENT_DOWNLOAD_COUNT,
1);
+
verify(mockServerMetrics).setValueOfGlobalGauge(eq(ServerGauge.PEER_DOWNLOAD_SPEED_MBPS),
eq(0L));
+ verifyNoMoreInteractions(mockServerMetrics);
+ }
+ }
+}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadSchedulerTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadSchedulerTest.java
index a5dca3bd8b5..7432681d289 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadSchedulerTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadSchedulerTest.java
@@ -22,6 +22,7 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
@@ -29,6 +30,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.io.FileUtils;
import org.apache.helix.model.InstanceConfig;
@@ -39,6 +41,8 @@ import
org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.env.CommonsConfigurationUtils;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import org.testng.annotations.AfterClass;
@@ -46,11 +50,15 @@ import org.testng.annotations.Test;
import static org.apache.pinot.server.predownload.PredownloadTestUtil.*;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
public class PredownloadSchedulerTest {
@@ -299,7 +307,8 @@ public class PredownloadSchedulerTest {
}
@Test
- public void testPredownloadParallelismConfiguration() throws Exception {
+ public void testPredownloadParallelismConfiguration()
+ throws Exception {
// Test default parallelism (should use numProcessors * 3)
Map<String, Object> defaultProps = Map.of(
"pinot.server.instance.id", INSTANCE_ID,
@@ -355,4 +364,423 @@ public class PredownloadSchedulerTest {
"Zero parallelism should fall back to default");
zeroScheduler.stop();
}
+
+ // ── Peer download tests
────────────────────────────────────────────────────
+
+ private PredownloadScheduler
buildPeerEnabledScheduler(PropertiesConfiguration properties)
+ throws Exception {
+
properties.setProperty(CommonConstants.Server.CONFIG_OF_PEER_DOWNLOAD_ENABLED,
true);
+ PredownloadScheduler scheduler = spy(new PredownloadScheduler(properties));
+ scheduler._executor = Runnable::run;
+
+ Field schemeField =
PredownloadScheduler.class.getDeclaredField("_peerDownloadScheme");
+ schemeField.setAccessible(true);
+ schemeField.set(scheduler, "http");
+
+ Field metricsField =
PredownloadScheduler.class.getDeclaredField("_predownloadMetrics");
+ metricsField.setAccessible(true);
+ metricsField.set(scheduler, mock(PredownloadMetrics.class));
+
+ return scheduler;
+ }
+
+ private void injectMockZkClient(PredownloadScheduler scheduler,
PredownloadZKClient mockZkClient)
+ throws Exception {
+ Field zkField =
PredownloadScheduler.class.getDeclaredField("_predownloadZkClient");
+ zkField.setAccessible(true);
+ zkField.set(scheduler, mockZkClient);
+ }
+
+ private void injectSegmentState(PredownloadScheduler scheduler,
+ List<PredownloadSegmentInfo> segments, Map<String, PredownloadTableInfo>
tableInfoMap)
+ throws Exception {
+ Field segField =
PredownloadScheduler.class.getDeclaredField("_predownloadSegmentInfoList");
+ segField.setAccessible(true);
+ segField.set(scheduler, segments);
+
+ Field mapField =
PredownloadScheduler.class.getDeclaredField("_tableInfoMap");
+ mapField.setAccessible(true);
+ mapField.set(scheduler, tableInfoMap);
+ }
+
+ @Test
+ public void testPeerDownloadEnabledViaConstructor()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+
+ Field enabledField =
PredownloadScheduler.class.getDeclaredField("_peerDownloadEnabled");
+ enabledField.setAccessible(true);
+
+ PredownloadScheduler disabled = new PredownloadScheduler(properties);
+ assertEquals(enabledField.get(disabled), false);
+ disabled.stop();
+
+
properties.setProperty(CommonConstants.Server.CONFIG_OF_PEER_DOWNLOAD_ENABLED,
true);
+ PredownloadScheduler enabled = new PredownloadScheduler(properties);
+ assertEquals(enabledField.get(enabled), true);
+ enabled.stop();
+ }
+
+ @Test
+ public void testDownloadSegmentFallbackToPeerOnDeepStoreFailure()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+ PredownloadZKClient mockZkClient = mock(PredownloadZKClient.class);
+ when(mockZkClient.getPeerServerURIs(any(), anyString(), anyString(),
anyString())).thenReturn(new ArrayList<>());
+ injectMockZkClient(scheduler, mockZkClient);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+ scheduler._failedSegments.add(SEGMENT_NAME);
+
+ File testFolder = new File(_temporaryFolder, "peerFallbackTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ when(_predownloadTableInfo.loadSegmentFromLocal(any())).thenReturn(false);
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ // 3-arg (deep store) throws — simulating exhausted deep store retries
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenThrow(new AttemptsExceededException("deep store failed", 3));
+ // 5-arg (peer) succeeds — supplier lambda is not invoked by mock, so no
ZK call needed
+ sfMock.when(
+ () -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(
+ anyString(), anyString(), any(), any(File.class),
anyString()))
+ .thenAnswer(inv -> null);
+
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ tarMock.when(() -> TarCompressionUtils.untar(any(File.class),
any(File.class)))
+ .thenAnswer(inv -> {
+ File untarDir = new File(testFolder, "untared_peer");
+ untarDir.mkdirs();
+ return List.of(untarDir);
+ });
+ scheduler.downloadSegment(segment);
+ }
+ }
+
+ assertFalse(scheduler._failedSegments.contains(SEGMENT_NAME),
+ "Segment should be removed from failed set after successful peer
download");
+ scheduler._executor = null; // lambda can't be cast to ThreadPoolExecutor
+ scheduler.stop();
+ }
+
+ @Test
+ public void testDownloadSegmentPeerDownloadStreamedOnDeepStoreFailure()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+
properties.setProperty("pinot.server.instance.segment.stream.download.untar",
true);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+ PredownloadZKClient mockZkClient = mock(PredownloadZKClient.class);
+ when(mockZkClient.getPeerServerURIs(any(), anyString(), anyString(),
anyString())).thenReturn(new ArrayList<>());
+ injectMockZkClient(scheduler, mockZkClient);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ SegmentZKMetadata zkMetadataWithoutCrypterName = createSegmentZKMetadata();
+ zkMetadataWithoutCrypterName.setCrypterName(null);
+ segment.updateSegmentInfo(zkMetadataWithoutCrypterName);
+ scheduler._failedSegments.add(SEGMENT_NAME);
+
+ File testFolder = new File(_temporaryFolder, "peerStreamedFallbackTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ when(_predownloadTableInfo.loadSegmentFromLocal(any())).thenReturn(false);
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ File untarDir = new File(testFolder, "untared_peer_streamed");
+ untarDir.mkdirs();
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ // deep store streamed (single-URI) download throws — simulating
exhausted deep store retries
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndStreamUntarToLocal(anyString(), any(File.class),
anyLong(),
+ any(AtomicInteger.class)))
+ .thenThrow(new AttemptsExceededException("deep store streamed
failed", 3));
+ // peer streamed (multi-peer supplier) download succeeds
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndStreamUntarToLocal(anyString(), anyString(),
any(), any(File.class),
+ anyLong()))
+ .thenReturn(untarDir);
+
+ scheduler.downloadSegment(segment);
+ }
+
+ assertFalse(scheduler._failedSegments.contains(SEGMENT_NAME),
+ "Segment should be removed from failed set after successful streamed
peer download");
+ scheduler._executor = null; // lambda can't be cast to ThreadPoolExecutor
+ scheduler.stop();
+ }
+
+ @Test
+ public void testDownloadSegmentNoPeerFallbackWhenPeerDisabled()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ // peerDownloadEnabled=false (default from properties) — no fallback to
peer
+ PredownloadScheduler scheduler = spy(new PredownloadScheduler(properties));
+ scheduler._executor = Runnable::run;
+ Field metricsField =
PredownloadScheduler.class.getDeclaredField("_predownloadMetrics");
+ metricsField.setAccessible(true);
+ metricsField.set(scheduler, mock(PredownloadMetrics.class));
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+
+ File testFolder = new File(_temporaryFolder, "noPeerTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenThrow(new AttemptsExceededException("deep store failed", 3));
+
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ assertThrows(AttemptsExceededException.class, () ->
scheduler.downloadSegment(segment));
+ }
+ }
+
+ assertTrue(scheduler._failedSegments.contains(SEGMENT_NAME),
+ "Segment should remain in failed set when peer download is disabled");
+ scheduler._executor = null;
+ scheduler.stop();
+ }
+
+ @Test
+ public void testDownloadSegmentBothDeepStoreAndPeerFail()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+ PredownloadZKClient mockZkClient = mock(PredownloadZKClient.class);
+ when(mockZkClient.getPeerServerURIs(any(), anyString(), anyString(),
anyString())).thenReturn(new ArrayList<>());
+ injectMockZkClient(scheduler, mockZkClient);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+
+ File testFolder = new File(_temporaryFolder, "bothFailTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenThrow(new AttemptsExceededException("deep store failed", 3));
+ sfMock.when(
+ () -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(
+ anyString(), anyString(), any(), any(File.class),
anyString()))
+ .thenThrow(new AttemptsExceededException("peer download failed", 3));
+
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ assertThrows(AttemptsExceededException.class, () ->
scheduler.downloadSegment(segment));
+ }
+ }
+
+ assertTrue(scheduler._failedSegments.contains(SEGMENT_NAME),
+ "Segment should be in failed set when both deep store and peer
download fail");
+ scheduler._executor = null;
+ scheduler.stop();
+ }
+
+ @Test
+ public void testDeepStoreSuccessEmitsDeepStoreMetric()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+
+ File testFolder = new File(_temporaryFolder, "deepStoreMetricTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ when(_predownloadTableInfo.loadSegmentFromLocal(any())).thenReturn(false);
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ PredownloadMetrics mockMetrics = getMockMetrics(scheduler);
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenAnswer(inv -> null);
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ tarMock.when(() -> TarCompressionUtils.untar(any(File.class),
any(File.class)))
+ .thenAnswer(inv -> {
+ File untarDir = new File(testFolder, "untared");
+ untarDir.mkdirs();
+ return List.of(untarDir);
+ });
+ scheduler.downloadSegment(segment);
+ }
+ }
+
+ verify(mockMetrics).deepStoreSegmentDownloaded();
+ verify(mockMetrics).segmentDownloaded(eq(true), eq(SEGMENT_NAME),
anyLong(), anyLong());
+ verify(mockMetrics, never()).peerSegmentDownloaded(anyBoolean(),
anyString(), anyLong(), anyLong());
+ scheduler._executor = null;
+ scheduler.stop();
+ }
+
+ @Test
+ public void testPeerDownloadSuccessEmitsPeerMetrics()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+ PredownloadZKClient mockZkClient = mock(PredownloadZKClient.class);
+ when(mockZkClient.getPeerServerURIs(any(), anyString(), anyString(),
anyString())).thenReturn(new ArrayList<>());
+ injectMockZkClient(scheduler, mockZkClient);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+
+ File testFolder = new File(_temporaryFolder, "peerMetricTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ when(_predownloadTableInfo.loadSegmentFromLocal(any())).thenReturn(false);
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ PredownloadMetrics mockMetrics = getMockMetrics(scheduler);
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenThrow(new AttemptsExceededException("deep store failed", 3));
+ sfMock.when(
+ () -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(
+ anyString(), anyString(), any(), any(File.class),
anyString()))
+ .thenAnswer(inv -> null);
+
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ tarMock.when(() -> TarCompressionUtils.untar(any(File.class),
any(File.class)))
+ .thenAnswer(inv -> {
+ File untarDir = new File(testFolder, "untared_peer");
+ untarDir.mkdirs();
+ return List.of(untarDir);
+ });
+ scheduler.downloadSegment(segment);
+ }
+ }
+
+ verify(mockMetrics, never()).deepStoreSegmentDownloaded();
+ verify(mockMetrics).peerSegmentDownloaded(eq(true), eq(SEGMENT_NAME),
anyLong(), anyLong());
+ verify(mockMetrics).segmentDownloaded(eq(true), eq(SEGMENT_NAME),
anyLong(), anyLong());
+ scheduler._executor = null;
+ scheduler.stop();
+ }
+
+ @Test
+ public void testPeerDownloadFailureEmitsSegmentLevelMetric()
+ throws Exception {
+ String propertiesFilePath =
+
this.getClass().getClassLoader().getResource(SAMPLE_PROPERTIES_FILE_NAME).getPath();
+ PropertiesConfiguration properties =
CommonsConfigurationUtils.fromPath(propertiesFilePath);
+ setUp(properties);
+
+ PredownloadScheduler scheduler = buildPeerEnabledScheduler(properties);
+ PredownloadZKClient mockZkClient = mock(PredownloadZKClient.class);
+ when(mockZkClient.getPeerServerURIs(any(), anyString(), anyString(),
anyString())).thenReturn(new ArrayList<>());
+ injectMockZkClient(scheduler, mockZkClient);
+
+ PredownloadSegmentInfo segment = new PredownloadSegmentInfo(TABLE_NAME,
SEGMENT_NAME);
+ segment.updateSegmentInfo(createSegmentZKMetadata());
+
+ File testFolder = new File(_temporaryFolder, "peerFailMetricTest");
+ testFolder.mkdirs();
+ String dataDir = testFolder.getAbsolutePath();
+ int lastIndex = dataDir.lastIndexOf(File.separator);
+
when(_predownloadTableInfo.getInstanceDataManagerConfig()).thenReturn(_instanceDataManagerConfig);
+ when(_predownloadTableInfo.getTableConfig()).thenReturn(_tableConfig);
+
when(_instanceDataManagerConfig.getInstanceDataDir()).thenReturn(dataDir.substring(0,
lastIndex));
+ when(_tableConfig.getTableName()).thenReturn(dataDir.substring(lastIndex +
1));
+ injectSegmentState(scheduler, List.of(segment), Map.of(TABLE_NAME,
_predownloadTableInfo));
+
+ PredownloadMetrics mockMetrics = getMockMetrics(scheduler);
+
+ try (MockedStatic<SegmentFetcherFactory> sfMock =
mockStatic(SegmentFetcherFactory.class)) {
+ sfMock.when(
+ () ->
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(anyString(),
any(File.class), anyString()))
+ .thenThrow(new AttemptsExceededException("deep store failed", 3));
+ sfMock.when(
+ () -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(
+ anyString(), anyString(), any(), any(File.class),
anyString()))
+ .thenThrow(new AttemptsExceededException("peer download failed", 3));
+
+ try (MockedStatic<TarCompressionUtils> tarMock =
mockStatic(TarCompressionUtils.class)) {
+ assertThrows(AttemptsExceededException.class, () ->
scheduler.downloadSegment(segment));
+ }
+ }
+
+ verify(mockMetrics, never()).deepStoreSegmentDownloaded();
+ verify(mockMetrics).peerSegmentDownloaded(eq(false), eq(SEGMENT_NAME),
eq(0L), eq(0L));
+ verify(mockMetrics).segmentDownloaded(eq(false), eq(SEGMENT_NAME), eq(0L),
eq(0L));
+ scheduler._executor = null;
+ scheduler.stop();
+ }
+
+ private PredownloadMetrics getMockMetrics(PredownloadScheduler scheduler)
+ throws Exception {
+ Field metricsField =
PredownloadScheduler.class.getDeclaredField("_predownloadMetrics");
+ metricsField.setAccessible(true);
+ return (PredownloadMetrics) metricsField.get(scheduler);
+ }
}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadZKClientTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadZKClientTest.java
index fda970a4272..3867c1927d7 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadZKClientTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/predownload/PredownloadZKClientTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.server.predownload;
+import java.net.URI;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -25,15 +26,20 @@ import java.util.Map;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.PropertyKey;
+import org.apache.helix.PropertyType;
import org.apache.helix.model.CurrentState;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.zookeeper.api.client.HelixZkClient;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
import org.mockito.MockedStatic;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -41,6 +47,7 @@ import org.testng.annotations.Test;
import static org.apache.pinot.server.predownload.PredownloadTestUtil.*;
import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.spy;
@@ -182,4 +189,69 @@ public class PredownloadZKClientTest {
assertEquals(predownloadSegmentInfoList.get(1).getCrc(), 0);
}
}
+
+ @Test
+ public void testGetPeerServerURIs()
+ throws Exception {
+ // Pass accessor directly — no spy needed since method signature accepts
accessor as param
+ HelixDataAccessor dataAccessor = mock(HelixDataAccessor.class);
+ when(dataAccessor.keyBuilder()).thenReturn(new
PropertyKey.Builder(CLUSTER_NAME));
+
+ ExternalView externalView = mock(ExternalView.class);
+ String peerInstanceId = "Server_peer_8098";
+ InstanceConfig peerConfig = mock(InstanceConfig.class);
+ when(peerConfig.getHostName()).thenReturn("peerHost");
+ ZNRecord peerRecord = new ZNRecord(peerInstanceId);
+ peerRecord.setIntField(CommonConstants.Helix.Instance.ADMIN_PORT_KEY,
9000);
+ when(peerConfig.getRecord()).thenReturn(peerRecord);
+
+ // Case 1: ExternalView is null → empty list
+ doAnswer(inv ->
null).when(dataAccessor).getProperty(any(PropertyKey.class));
+ assertTrue(_predownloadZkClient.getPeerServerURIs(dataAccessor,
TABLE_NAME, SEGMENT_NAME, "http").isEmpty(),
+ "Should return empty list when ExternalView is null");
+
+ // Case 2: ExternalView found but segment has no state map → empty list
+ doAnswer(inv -> {
+ PropertyKey key = inv.getArgument(0);
+ return (key.getType() == PropertyType.EXTERNALVIEW) ? externalView :
null;
+ }).when(dataAccessor).getProperty(any(PropertyKey.class));
+ when(externalView.getStateMap(SEGMENT_NAME)).thenReturn(null);
+ assertTrue(_predownloadZkClient.getPeerServerURIs(dataAccessor,
TABLE_NAME, SEGMENT_NAME, "http").isEmpty(),
+ "Should return empty list when segment not in ExternalView");
+
+ // Case 3: Only self ONLINE → empty list
+ // In @BeforeClass: new PredownloadZKClient(ZK_ADDRESS, INSTANCE_ID,
CLUSTER_NAME)
+ // → _instanceName = CLUSTER_NAME
+
when(externalView.getStateMap(SEGMENT_NAME)).thenReturn(Map.of(CLUSTER_NAME,
"ONLINE"));
+ assertTrue(_predownloadZkClient.getPeerServerURIs(dataAccessor,
TABLE_NAME, SEGMENT_NAME, "http").isEmpty(),
+ "Should exclude self instance from peer list");
+
+ // Case 4: peer ONLINE + one OFFLINE → only peer returned (self-exclusion
tested in Case 3)
+ Map<String, String> stateMap = new HashMap<>();
+ stateMap.put(peerInstanceId, "ONLINE"); // peer → included
+ stateMap.put("Server_offline", "OFFLINE"); // offline → excluded
+ when(externalView.getStateMap(SEGMENT_NAME)).thenReturn(stateMap);
+ doAnswer(inv -> {
+ PropertyKey key = inv.getArgument(0);
+ return (key.getType() == PropertyType.EXTERNALVIEW) ? externalView :
peerConfig;
+ }).when(dataAccessor).getProperty(any(PropertyKey.class));
+
+ List<URI> uris = _predownloadZkClient.getPeerServerURIs(dataAccessor,
TABLE_NAME, SEGMENT_NAME, "http");
+ assertEquals(uris.size(), 1, "Should return exactly one peer URI");
+ String uriStr = uris.get(0).toString();
+ assertTrue(uriStr.contains("peerHost"), "URI should contain peer
hostname");
+ assertTrue(uriStr.contains("9000"), "URI should contain peer admin port");
+ assertTrue(uriStr.contains(TABLE_NAME), "URI should contain table name");
+ assertTrue(uriStr.contains(SEGMENT_NAME), "URI should contain segment
name");
+ assertTrue(uriStr.startsWith("http://";), "URI should use http scheme");
+
+ // Case 5: Peer InstanceConfig is null → skipped, empty list
+
when(externalView.getStateMap(SEGMENT_NAME)).thenReturn(Map.of(peerInstanceId,
"ONLINE"));
+ doAnswer(inv -> {
+ PropertyKey key = inv.getArgument(0);
+ return (key.getType() == PropertyType.EXTERNALVIEW) ? externalView :
null;
+ }).when(dataAccessor).getProperty(any(PropertyKey.class));
+ assertTrue(_predownloadZkClient.getPeerServerURIs(dataAccessor,
TABLE_NAME, SEGMENT_NAME, "http").isEmpty(),
+ "Should skip peer when its InstanceConfig is null");
+ }
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index b81f4fb2828..ee25435988a 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -1727,6 +1727,8 @@ public class CommonConstants {
// Predownload related configs
public static final String CONFIG_OF_PREDOWNLOAD_PARALLELISM =
"pinot.server.predownload.parallelism";
public static final int DEFAULT_PREDOWNLOAD_PARALLELISM = -1; // Use
numProcessors * 3 as default
+ public static final String CONFIG_OF_PEER_DOWNLOAD_ENABLED =
"pinot.server.peer.download.enabled";
+ public static final boolean DEFAULT_PEER_DOWNLOAD_ENABLED = false;
public static final String CONFIG_OF_CURRENT_DATA_TABLE_VERSION =
"pinot.server.instance.currentDataTableVersion";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
