You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by jl...@apache.org on 2020/05/13 20:38:25 UTC
[incubator-pinot] branch master updated: Fix PerfBenchmarkDriver on
uploading/downloading segments (#5381)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 8f9a2f6 Fix PerfBenchmarkDriver on uploading/downloading segments (#5381)
8f9a2f6 is described below
commit 8f9a2f6c276a91c3c2caa3478bd4e7b39e4bc217
Author: Jialiang Li <jl...@linkedin.com>
AuthorDate: Wed May 13 13:38:12 2020 -0700
Fix PerfBenchmarkDriver on uploading/downloading segments (#5381)
Co-authored-by: Jack Li(Analytics Engineering) <jl...@jlli-mn1.linkedin.biz>
---
.../apache/pinot/controller/api/upload/ZKOperator.java | 7 ++++---
.../java/org/apache/pinot/perf/BenchmarkQueryEngine.java | 2 +-
.../org/apache/pinot/tools/perf/PerfBenchmarkDriver.java | 15 ++++++++++++---
.../org/apache/pinot/tools/perf/PerfBenchmarkRunner.java | 2 +-
4 files changed, 18 insertions(+), 8 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
index b33161b..b126605 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
@@ -20,6 +20,7 @@ package org.apache.pinot.controller.api.upload;
import java.io.File;
import java.net.URI;
+import javax.annotation.Nullable;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response;
import org.apache.helix.ZNRecord;
@@ -59,7 +60,7 @@ public class ZKOperator {
public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
- HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation)
+ @Nullable HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation)
throws Exception {
String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName);
String segmentName = segmentMetadata.getName();
@@ -69,7 +70,7 @@ public class ZKOperator {
_pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName);
if (segmentMetadataZnRecord == null) {
LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName);
- String crypter = headers.getHeaderString(FileUploadDownloadClient.CustomHeaders.CRYPTER);
+ String crypter = headers != null ? headers.getHeaderString(FileUploadDownloadClient.CustomHeaders.CRYPTER) : null;
processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter,
rawTableName, segmentName, moveSegmentToFinalLocation);
return;
@@ -202,7 +203,7 @@ public class ZKOperator {
}
private void processNewSegment(SegmentMetadata segmentMetadata, URI finalSegmentLocationURI,
- File currentSegmentLocation, String zkDownloadURI, String crypter, String rawTableName, String segmentName,
+ File currentSegmentLocation, String zkDownloadURI, @Nullable String crypter, String rawTableName, String segmentName,
boolean moveSegmentToFinalLocation) {
// For v1 segment uploads, we will not move the segment
if (moveSegmentToFinalLocation) {
diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkQueryEngine.java b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkQueryEngine.java
index ce9c446..1f447a0 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkQueryEngine.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkQueryEngine.java
@@ -104,7 +104,7 @@ public class BenchmarkQueryEngine {
SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(segmentDir);
_perfBenchmarkDriver.configureTable(TABLE_NAME);
System.out.println("Adding segment " + segmentDir.getAbsolutePath());
- _perfBenchmarkDriver.addSegment(TABLE_NAME, segmentMetadata);
+ _perfBenchmarkDriver.addSegment(TABLE_NAME, segmentDir, segmentMetadata);
}
ZkClient client = new ZkClient("localhost:2191", 10000, 10000, new ZNRecordSerializer());
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
index cba0555..d953e86 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
@@ -28,6 +28,7 @@ import java.io.FileInputStream;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
+import java.net.URI;
import java.net.URL;
import java.net.URLConnection;
import java.nio.charset.StandardCharsets;
@@ -45,9 +46,12 @@ import org.apache.helix.manager.zk.ZKHelixAdmin;
import org.apache.helix.tools.ClusterVerifiers.StrictMatchExternalViewVerifier;
import org.apache.pinot.broker.broker.helix.HelixBrokerStarter;
import org.apache.pinot.common.utils.CommonConstants;
+import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
+import org.apache.pinot.controller.api.resources.ControllerFilePathProvider;
+import org.apache.pinot.controller.api.upload.ZKOperator;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.core.segment.index.metadata.SegmentMetadata;
import org.apache.pinot.server.starter.helix.HelixServerStarter;
@@ -322,10 +326,15 @@ public class PerfBenchmarkDriver {
*
* @param segmentMetadata segment metadata.
*/
- public void addSegment(String tableName, SegmentMetadata segmentMetadata) {
+ public void addSegment(String tableName, File segmentFile, SegmentMetadata segmentMetadata) throws Exception {
String rawTableName = TableNameBuilder.extractRawTableName(tableName);
- _helixResourceManager
- .addNewSegment(rawTableName, segmentMetadata, "http://" + _controllerAddress + "/" + segmentMetadata.getName());
+ URI finalSegmentLocationURI = URIUtils
+ .getUri(ControllerFilePathProvider.getInstance().getDataDirURI().toString(), rawTableName,
+ URIUtils.encode(segmentMetadata.getName()));
+ String zkDownloadURI = "http://" + _controllerAddress + "/segments/" + rawTableName + "/" + segmentMetadata.getName();
+ ZKOperator zkOperator = new ZKOperator(_helixResourceManager, getControllerConf(), null);
+ zkOperator.completeSegmentOperations(rawTableName, segmentMetadata, finalSegmentLocationURI, segmentFile,
+ false, null, zkDownloadURI, true);
}
public static void waitForExternalViewUpdate(String zkAddress, final String clusterName, long timeoutInMilliseconds) {
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkRunner.java b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkRunner.java
index 37f2147..959826b 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkRunner.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkRunner.java
@@ -166,7 +166,7 @@ public class PerfBenchmarkRunner extends AbstractBaseCommand implements Command
driver.configureTable(tableName, invertedIndexColumns, bloomFilterColumns);
tableConfigured = true;
}
- driver.addSegment(tableName, segmentMetadata);
+ driver.addSegment(tableName, segment, segmentMetadata);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org