You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ti...@apache.org on 2022/11/22 16:39:22 UTC
[pinot] branch master updated: Download offline segments from peers (#9710)
This is an automated email from the ASF dual-hosted git repository.
tingchen 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 6cfa6dca8b Download offline segments from peers (#9710)
6cfa6dca8b is described below
commit 6cfa6dca8b0c5b38f2d4c3684e43aea3ad7242a9
Author: Xuanyi Li <sh...@outlook.com>
AuthorDate: Tue Nov 22 08:39:14 2022 -0800
Download offline segments from peers (#9710)
* download offline segments from peers when segment streaming download is disabled
* fix checkstyle
* add metrics for remote and peer download failure
* throw exception when fail to use peers as the last resort
* fix the check style
* add unit test of baseTableDataManager
* unit test the overriding of peerDownLoad scheme config
* fix the error when get peers URI by directly passing tableNameWithType
* cosmetic change of unit test
---
.../apache/pinot/common/metrics/ServerMeter.java | 2 +
.../utils/fetcher/SegmentFetcherFactory.java | 27 ++++++
pinot-core/pom.xml | 2 +-
.../core/data/manager/BaseTableDataManager.java | 42 +++++++++-
.../pinot/core/util/PeerServerSegmentFinder.java | 4 +
.../data/manager/BaseTableDataManagerTest.java | 96 +++++++++++++++++++++-
.../local/data/manager/TableDataManagerConfig.java | 20 +++++
.../data/manager/TableDataManagerConfigTest.java | 54 ++++++++++++
.../helix/HelixInstanceDataManagerConfig.java | 6 ++
.../config/instance/InstanceDataManagerConfig.java | 2 +
10 files changed, 248 insertions(+), 7 deletions(-)
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 e086545c1e..65042f1b52 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
@@ -71,6 +71,8 @@ public enum ServerMeter implements AbstractMetrics.Meter {
SEGMENT_STREAMED_DOWNLOAD_UNTAR_FAILURES("segments", false),
SEGMENT_DIR_MOVEMENT_FAILURES("segments", false),
SEGMENT_DOWNLOAD_FAILURES("segments", false),
+ SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES("segments", false),
+ SEGMENT_DOWNLOAD_FROM_PEERS_FAILURES("segments", false),
NUM_RESIZES("numResizes", false),
NO_TABLE_ACCESS("tables", true),
INDEXING_FAILURES("attributeValues", true),
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 eb5b498729..2fc0124cb4 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
@@ -18,18 +18,21 @@
*/
package org.apache.pinot.common.utils.fetcher;
+import com.google.common.base.Preconditions;
import java.io.File;
import java.net.URI;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import org.apache.pinot.common.auth.AuthConfig;
import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.spi.crypt.PinotCrypter;
import org.apache.pinot.spi.crypt.PinotCrypterFactory;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.checkerframework.checker.nullness.qual.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,6 +46,7 @@ public class SegmentFetcherFactory {
private static final String AUTH_KEY = CommonConstants.KEY_OF_AUTH;
private static final Logger LOGGER = LoggerFactory.getLogger(SegmentFetcherFactory.class);
+ private static final Random RANDOM = new Random();
private final Map<String, SegmentFetcher> _segmentFetcherMap = new HashMap<>();
private final SegmentFetcher _httpSegmentFetcher = new HttpSegmentFetcher();
@@ -182,6 +186,12 @@ public class SegmentFetcherFactory {
getInstance().fetchAndDecryptSegmentToLocalInternal(uri, dest, crypterName);
}
+ // uris have equal weight to be selected for segment download
+ public static void fetchAndDecryptSegmentToLocal(List<URI> uris, File dest, String crypterName)
+ throws Exception {
+ getInstance().fetchAndDecryptSegmentToLocalInternal(uris, dest, crypterName);
+ }
+
private void fetchAndDecryptSegmentToLocalInternal(String uri, File dest, String crypterName)
throws Exception {
if (crypterName == null) {
@@ -196,4 +206,21 @@ public class SegmentFetcherFactory {
crypter.decrypt(tempDownloadedFile, dest);
}
}
+
+ private void fetchAndDecryptSegmentToLocalInternal(@NonNull List<URI> uris, File dest, String crypterName)
+ throws Exception {
+ Preconditions.checkArgument(!uris.isEmpty(), "empty uris passed into the fetchAndDecryptSegmentToLocalInternal");
+ URI uri = uris.get(RANDOM.nextInt(uris.size()));
+ if (crypterName == null) {
+ fetchSegmentToLocal(uri, dest);
+ } else {
+ // download
+ File tempDownloadedFile = new File(dest.getPath() + ENCODED_SUFFIX);
+ fetchSegmentToLocal(uri, tempDownloadedFile);
+
+ // decrypt
+ PinotCrypter crypter = PinotCrypterFactory.create(crypterName);
+ crypter.decrypt(tempDownloadedFile, dest);
+ }
+ }
}
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index c140be397b..fe7aaee114 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -218,7 +218,7 @@
</dependency>
<dependency>
<groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
+ <artifactId>mockito-inline</artifactId>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
index 6b37af6c8b..b6386375a6 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
@@ -25,6 +25,7 @@ import com.google.common.cache.CacheBuilder;
import com.google.common.cache.LoadingCache;
import java.io.File;
import java.io.IOException;
+import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -53,6 +54,7 @@ import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.common.utils.config.TierConfigUtils;
import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager;
+import org.apache.pinot.core.util.PeerServerSegmentFinder;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManagerConfig;
@@ -511,6 +513,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
throws Exception {
File tarFile = new File(tempRootDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
String uri = zkMetadata.getDownloadUrl();
+ boolean downloadSuccess = false;
try {
if (_segmentDownloadSemaphore != null) {
long startTime = System.currentTimeMillis();
@@ -523,19 +526,54 @@ public abstract class BaseTableDataManager implements TableDataManager {
SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(uri, tarFile, zkMetadata.getCrypterName());
LOGGER.info("Downloaded tarred segment: {} for table: {} from: {} to: {}, file length: {}", segmentName,
_tableNameWithType, uri, tarFile, tarFile.length());
+ downloadSuccess = true;
return tarFile;
} catch (AttemptsExceededException e) {
LOGGER.error("Attempts exceeded when downloading segment: {} for table: {} from: {} to: {}", segmentName,
_tableNameWithType, uri, tarFile);
- _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FAILURES, 1L);
- throw e;
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES, 1L);
+ if (_tableDataManagerConfig.getTablePeerDownloadScheme() == null) {
+ throw e;
+ }
+ downloadFromPeersWithoutStreaming(segmentName, zkMetadata, tarFile);
+ downloadSuccess = true;
+ return tarFile;
} finally {
+ if (!downloadSuccess) {
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FAILURES, 1L);
+ }
if (_segmentDownloadSemaphore != null) {
_segmentDownloadSemaphore.release();
}
}
}
+ // not thread safe. Caller should invoke it with safe concurrency control.
+ protected void downloadFromPeersWithoutStreaming(String segmentName, SegmentZKMetadata zkMetadata,
+ File destTarFile) throws Exception {
+ Preconditions.checkArgument(_tableDataManagerConfig.getTablePeerDownloadScheme() != null,
+ "Download peers require non null peer download scheme");
+ List<URI> peerSegmentURIs = PeerServerSegmentFinder.getPeerServerURIs(segmentName,
+ _tableDataManagerConfig.getTablePeerDownloadScheme(), _helixManager, _tableNameWithType);
+ if (peerSegmentURIs.isEmpty()) {
+ String msg = String.format("segment %s doesn't have any peers", segmentName);
+ LOGGER.warn(msg);
+ // HelixStateTransitionHandler would catch the runtime exception and mark the segment state as Error
+ throw new RuntimeException(msg);
+ }
+ try {
+ // Next download the segment from a randomly chosen server using configured scheme.
+ SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(peerSegmentURIs, destTarFile, zkMetadata.getCrypterName());
+ LOGGER.info("Fetched segment {} from peers: {} to: {} of size: {}", segmentName, peerSegmentURIs, destTarFile,
+ destTarFile.length());
+ } catch (AttemptsExceededException e) {
+ LOGGER.error("Attempts exceeded when downloading segment: {} for table: {} from peers {} to: {}", segmentName,
+ _tableNameWithType, peerSegmentURIs, destTarFile);
+ _serverMetrics.addMeteredTableValue(_tableNameWithType, ServerMeter.SEGMENT_DOWNLOAD_FROM_PEERS_FAILURES, 1L);
+ throw e;
+ }
+ }
+
private File downloadAndStreamUntarWithRateLimit(String segmentName, SegmentZKMetadata zkMetadata, File tempRootDir,
long maxStreamRateInByte)
throws Exception {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/PeerServerSegmentFinder.java b/pinot-core/src/main/java/org/apache/pinot/core/util/PeerServerSegmentFinder.java
index 727a229d3e..80bd31fa34 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/PeerServerSegmentFinder.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/PeerServerSegmentFinder.java
@@ -64,7 +64,11 @@ public class PeerServerSegmentFinder {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
String tableNameWithType =
TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(llcSegmentName.getTableName());
+ return getPeerServerURIs(segmentName, downloadScheme, helixManager, tableNameWithType);
+ }
+ public static List<URI> getPeerServerURIs(String segmentName, String downloadScheme,
+ HelixManager helixManager, String tableNameWithType) {
HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
String clusterName = helixManager.getClusterName();
if (clusterName == null) {
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
index 62b55df352..6437a54a70 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
@@ -20,6 +20,8 @@ package org.apache.pinot.core.data.manager;
import java.io.File;
import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -38,6 +40,7 @@ import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.common.utils.fetcher.BaseSegmentFetcher;
import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager;
+import org.apache.pinot.core.util.PeerServerSegmentFinder;
import org.apache.pinot.segment.local.data.manager.TableDataManagerConfig;
import org.apache.pinot.segment.local.data.manager.TableDataManagerParams;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
@@ -62,11 +65,17 @@ import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.retry.AttemptsExceededException;
import org.apache.pinot.util.TestUtils;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -614,6 +623,53 @@ public class BaseTableDataManagerTest {
}
}
+ // case 2: if the attempt to download from deep storage exceeds, invoke downloadFromPeers.
+ @Test
+ public void testDownloadAndDecryptPeerDownload() throws Exception {
+ String backupCopyURI = mockRemoteCopy().toString();
+ SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
+ when(zkmd.getDownloadUrl()).thenReturn(backupCopyURI);
+
+ TableDataManagerConfig config = createDefaultTableDataManagerConfig();
+ when(config.getTablePeerDownloadScheme()).thenReturn("http");
+ BaseTableDataManager tmgr = createSpyOfflineTableManager(config);
+ File tempRootDir = tmgr.getTmpSegmentDataDir("test-download-decrypt-peer");
+
+ // As the case 2 description says, we need to mock the static method fetchAndDecryptSegmentToLocal to
+ // throw the AttemptExceed exception; Due to the constraint that mockito static cannot do argument matching,
+ // e.g., any(), we have to pass exact argument value when mocking fetchAndDecryptSegmentToLocal.
+ // However, the second argument of File is internally created, which cannot be mocked.
+ // Luckily, the File class's equal method only compares the path. Thus, we can create a file with identical path
+ // and use it to mock the fetchAndDecryptSegmentToLocal
+ File destFile = new File(tempRootDir, "seg01" + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ doNothing().when(tmgr).downloadFromPeersWithoutStreaming("seg01", zkmd, destFile);
+ try (MockedStatic<SegmentFetcherFactory> mockSegFactory = mockStatic(SegmentFetcherFactory.class)) {
+ mockSegFactory.when(() -> SegmentFetcherFactory.fetchAndDecryptSegmentToLocal(backupCopyURI, destFile, null))
+ .thenThrow(new AttemptsExceededException("fake attempt exceeds exception"));
+ tmgr.downloadAndDecrypt("seg01", zkmd, tempRootDir);
+ }
+ verify(tmgr, times(1)).downloadFromPeersWithoutStreaming("seg01", zkmd, destFile);
+ }
+
+ // happy case: download from peers
+ @Test
+ public void testDownloadFromPeersWithoutStreaming() throws Exception {
+ URI uri = mockRemoteCopy();
+ TableDataManagerConfig config = createDefaultTableDataManagerConfig();
+ when(config.getTablePeerDownloadScheme()).thenReturn("http");
+ HelixManager mockedHelix = mock(HelixManager.class);
+ BaseTableDataManager tmgr = createTableManager(config, mockedHelix);
+ File tempRootDir = tmgr.getTmpSegmentDataDir("test-download-peer-without-streaming");
+ File destFile = new File(tempRootDir, "seg01" + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ try (MockedStatic<PeerServerSegmentFinder> mockPeerSegFinder = mockStatic(PeerServerSegmentFinder.class)) {
+ mockPeerSegFinder.when(() -> PeerServerSegmentFinder.getPeerServerURIs(
+ "seg01", "http", mockedHelix, TABLE_NAME_WITH_TYPE))
+ .thenReturn(Collections.singletonList(uri));
+ tmgr.downloadFromPeersWithoutStreaming("seg01", mock(SegmentZKMetadata.class), destFile);
+ }
+ assertEquals(FileUtils.readFileToString(destFile), "this is from somewhere remote");
+ }
+
@Test
public void testUntarAndMoveSegment()
throws IOException {
@@ -661,10 +717,7 @@ public class BaseTableDataManagerTest {
}
private static BaseTableDataManager createTableManager() {
- TableDataManagerConfig config = mock(TableDataManagerConfig.class);
- when(config.getTableName()).thenReturn(TABLE_NAME_WITH_TYPE);
- when(config.getDataDir()).thenReturn(TABLE_DATA_DIR.getAbsolutePath());
- when(config.getAuthConfig()).thenReturn(new MapConfiguration(Collections.emptyMap()));
+ TableDataManagerConfig config = createDefaultTableDataManagerConfig();
OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
tableDataManager.init(config, "dummyInstance", mock(ZkHelixPropertyStore.class),
@@ -674,6 +727,32 @@ public class BaseTableDataManagerTest {
return tableDataManager;
}
+ private static BaseTableDataManager createTableManager(TableDataManagerConfig config, HelixManager helixManager) {
+ OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
+ tableDataManager.init(config, "dummyInstance", mock(ZkHelixPropertyStore.class),
+ new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()), helixManager, null,
+ new TableDataManagerParams(0, false, -1));
+ tableDataManager.start();
+ return tableDataManager;
+ }
+
+ private static OfflineTableDataManager createSpyOfflineTableManager(TableDataManagerConfig tableDataManagerConfig) {
+ OfflineTableDataManager tableDataManager = new OfflineTableDataManager();
+ tableDataManager.init(tableDataManagerConfig, "dummyInstance", mock(ZkHelixPropertyStore.class),
+ new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()), mock(HelixManager.class), null,
+ new TableDataManagerParams(0, false, -1));
+ tableDataManager.start();
+ return Mockito.spy(tableDataManager);
+ }
+
+ private static TableDataManagerConfig createDefaultTableDataManagerConfig() {
+ TableDataManagerConfig config = mock(TableDataManagerConfig.class);
+ when(config.getTableName()).thenReturn(TABLE_NAME_WITH_TYPE);
+ when(config.getDataDir()).thenReturn(TABLE_DATA_DIR.getAbsolutePath());
+ when(config.getAuthConfig()).thenReturn(new MapConfiguration(Collections.emptyMap()));
+ return config;
+ }
+
private static SegmentZKMetadata createRawSegment(TableConfig tableConfig, String segName, SegmentVersion segVer,
int rowCnt)
throws Exception {
@@ -719,4 +798,13 @@ public class BaseTableDataManagerTest {
TierFactory.PINOT_SERVER_STORAGE_TYPE, "tag_OFFLINE", null,
Collections.singletonMap("dataDir", dataDir.getAbsolutePath())))).build();
}
+
+ private static URI mockRemoteCopy() throws IOException, URISyntaxException {
+ File tempInput = new File(TEMP_DIR, "tmp.txt");
+ FileUtils.write(tempInput, "this is from somewhere remote");
+
+ String backupCopyURI = "file://" + tempInput.getAbsolutePath();
+ URI uri = new URI(backupCopyURI);
+ return uri;
+ }
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfig.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfig.java
index da1f5efc71..fa27c67132 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfig.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfig.java
@@ -22,15 +22,20 @@ import com.google.common.base.Preconditions;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The config used for TableDataManager.
*/
public class TableDataManagerConfig {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TableDataManagerConfig.class);
+
private static final String TABLE_DATA_MANAGER_TYPE = "dataManagerType";
private static final String TABLE_DATA_MANAGER_DATA_DIRECTORY = "directory";
private static final String TABLE_DATA_MANAGER_CONSUMER_DIRECTORY = "consumerDirectory";
@@ -39,6 +44,7 @@ public class TableDataManagerConfig {
private static final String TABLE_DATA_MANAGER_AUTH = "auth";
private static final String TABLE_DELETED_SEGMENTS_CACHE_SIZE = "deletedSegmentsCacheSize";
private static final String TABLE_DELETED_SEGMENTS_CACHE_TTL_MINUTES = "deletedSegmentsCacheTTL";
+ private static final String TABLE_PEER_DOWNLOAD_SCHEME = "peerDownloadScheme";
private final Configuration _tableDataManagerConfig;
@@ -82,6 +88,10 @@ public class TableDataManagerConfig {
return _tableDataManagerConfig.getInt(TABLE_DELETED_SEGMENTS_CACHE_TTL_MINUTES);
}
+ public String getTablePeerDownloadScheme() {
+ return _tableDataManagerConfig.getString(TABLE_PEER_DOWNLOAD_SCHEME);
+ }
+
public static TableDataManagerConfig getDefaultHelixTableDataManagerConfig(
InstanceDataManagerConfig instanceDataManagerConfig, String tableNameWithType) {
Configuration defaultConfig = new PropertiesConfiguration();
@@ -96,6 +106,11 @@ public class TableDataManagerConfig {
instanceDataManagerConfig.getDeletedSegmentsCacheSize());
defaultConfig.addProperty(TABLE_DELETED_SEGMENTS_CACHE_TTL_MINUTES,
instanceDataManagerConfig.getDeletedSegmentsCacheTtlMinutes());
+ // allow null
+ String segmentPeerDownloadScheme = instanceDataManagerConfig.getSegmentPeerDownloadScheme();
+ LOGGER.info("instance level segment peer download scheme = {}", segmentPeerDownloadScheme);
+ defaultConfig.addProperty(TABLE_PEER_DOWNLOAD_SCHEME, segmentPeerDownloadScheme);
+
// copy auth-related configs
instanceDataManagerConfig.getConfig().subset(TABLE_DATA_MANAGER_AUTH).toMap()
@@ -108,6 +123,11 @@ public class TableDataManagerConfig {
// Override table level configs
_tableDataManagerConfig.addProperty(TABLE_IS_DIMENSION, tableConfig.isDimTable());
+ SegmentsValidationAndRetentionConfig segmentConfig = tableConfig.getValidationConfig();
+ if (segmentConfig != null && segmentConfig.getPeerSegmentDownloadScheme() != null) {
+ _tableDataManagerConfig.setProperty(TABLE_PEER_DOWNLOAD_SCHEME, segmentConfig.getPeerSegmentDownloadScheme());
+ }
+ LOGGER.info("final segment peer download scheme = {}", getTablePeerDownloadScheme());
// If we wish to override some table level configs using table config, override them here
// Note: the configs in TableDataManagerConfig is immutable once the table is created, which mean it will not pick
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfigTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfigTest.java
new file mode 100644
index 0000000000..be42634e97
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/data/manager/TableDataManagerConfigTest.java
@@ -0,0 +1,54 @@
+/**
+ * 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.segment.local.data.manager;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+
+
+public class TableDataManagerConfigTest {
+
+ @Test
+ public void testOverridePeerDownloadScheme() {
+ // Use tableConfig's peer download scheme is present
+ Configuration defaultConfig = new PropertiesConfiguration();
+ defaultConfig.addProperty("peerDownloadScheme", "foo");
+ TableDataManagerConfig finalConfig = new TableDataManagerConfig(defaultConfig);
+ TableConfig tableConfig = mock(TableConfig.class);
+ SegmentsValidationAndRetentionConfig segConfig = mock(SegmentsValidationAndRetentionConfig.class);
+ when(tableConfig.getValidationConfig()).thenReturn(segConfig);
+ when(segConfig.getPeerSegmentDownloadScheme()).thenReturn("http");
+ finalConfig.overrideConfigs(tableConfig);
+ assertEquals("http", finalConfig.getTablePeerDownloadScheme());
+
+ // Use default value if tableConfig's peer download scheme is absent
+ finalConfig = new TableDataManagerConfig(new PropertiesConfiguration());
+ when(segConfig.getPeerSegmentDownloadScheme()).thenReturn(null);
+ finalConfig.overrideConfigs(tableConfig);
+ assertNull(finalConfig.getTablePeerDownloadScheme());
+ }
+}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java
index dd3a839f44..f5bfc87805 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java
@@ -123,6 +123,7 @@ public class HelixInstanceDataManagerConfig implements InstanceDataManagerConfig
private static final String DELETED_SEGMENTS_CACHE_SIZE = "table.deleted.segments.cache.size";
private static final String DELETED_SEGMENTS_CACHE_TTL_MINUTES = "table.deleted.segments.cache.ttl.minutes";
+ private static final String PEER_DOWNLOAD_SCHEME = "peer.download.scheme";
private final static String[] REQUIRED_KEYS = {INSTANCE_ID, INSTANCE_DATA_DIR, READ_MODE};
private static final long DEFAULT_ERROR_CACHE_SIZE = 100L;
@@ -274,6 +275,11 @@ public class HelixInstanceDataManagerConfig implements InstanceDataManagerConfig
DEFAULT_DELETED_SEGMENTS_CACHE_TTL_MINUTES);
}
+ @Override
+ public String getSegmentPeerDownloadScheme() {
+ return _instanceDataManagerConfiguration.getProperty(PEER_DOWNLOAD_SCHEME);
+ }
+
@Override
public String toString() {
String configString = "";
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/instance/InstanceDataManagerConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/instance/InstanceDataManagerConfig.java
index 514901490e..9069ee7a57 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/instance/InstanceDataManagerConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/instance/InstanceDataManagerConfig.java
@@ -66,4 +66,6 @@ public interface InstanceDataManagerConfig {
int getDeletedSegmentsCacheSize();
int getDeletedSegmentsCacheTtlMinutes();
+
+ String getSegmentPeerDownloadScheme();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org