You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2020/04/28 17:28:12 UTC
[GitHub] [incubator-pinot] chenboat opened a new pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
chenboat opened a new pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314
Add a new best effort segment uploader with bounded upload time and default store uri if upload fails. It is designed to enable by-passing the deep store requirement in LLC split commit.
Twig the SegmentUploader interface to add table name and segment name. The reason is to make the BestEffortSegmentUploader as a singleton for sharing in a Pinot server.
@mcvsubbu
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r420412763
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/Server2ControllerSegmentUploader.java
##########
@@ -24,6 +24,7 @@
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.common.utils.LLCSegmentName;
Review comment:
I meant that we shouldnot need the segment name argument in the constructor of the class. Sorry, I should have been clearer. You can do this in another PR. It is not a public interface, so we are ok
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418733908
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does segment upload to a segment store (with store root dir configured as
Review comment:
Can you put this in javadoc format?
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr) using PinotFS within a configurable timeout period. The final segment location would be in the
+// URI _segmentStoreUriStr/_tableNameWithType/segmentName if successful.
+public class PinotFSSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(PinotFSSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ private int _timeoutInMs;
+
+ public PinotFSSegmentUploader(String segmentStoreDirUri, int timeoutMillis) {
+ _segmentStoreUriStr = segmentStoreDirUri;
+ _timeoutInMs = timeoutMillis;
+ }
+
+ public URI uploadSegment(File segmentFile, LLCSegmentName segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return null;
+ }
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil
+ .join(File.separator, _segmentStoreUriStr, segmentName.getTableName(), segmentName.getSegmentName()));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.warn("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
Review comment:
include segment destUri in the log
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java
##########
@@ -35,16 +38,34 @@
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
private final IndexLoadingConfig _indexLoadingConfig;
private final SegmentUploader _segmentUploader;
+ // The default segment location uri str, could be null.
+ private final String _defaultSegmentLocation;
private final Logger _segmentLogger;
public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader) {
+ this(segmentLogger, protocolHandler, indexLoadingConfig, params, segmentUploader, null);
+ }
+
+ /**
+ *
+ * @param segmentLogger
+ * @param protocolHandler
+ * @param indexLoadingConfig
+ * @param params
+ * @param segmentUploader
+ * @param defaultSegmentLocation The default segment location uri str, could be null.
+ */
+ public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
+ IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader,
+ String defaultSegmentLocation) {
Review comment:
I think we can clean this up by not having the extra argument, but just taking TableConfig instead of indexLoadingConfig in the constructor. We can pull from tableconfig whether the default (always a peer). Construct the peer download URL in this class if the upload fails. For now, you can comment out that failure handling code.
Let us float a discussion on how the table should be configured on upload failure. You can either do it in doc, or via email or a discsussion in the chat channel (and summarize in email).
A few options come to mind, but best to discuss with the team before moving forward.
1. In the `SegmentValidationAndRetentionConfig` set some variable to indicate that we are enabling peer download for segments. The variable can just indicate the scheme: e.g. "http". The rest of the URL can be constructed easily since we know the value of the host, port. etc. I prefer this to having a boolean and then another one for the scheme somewhere else. Putting it here can also help in the case when deep store is down and we want to download offline table segments.
2. Introduce something in the streamConfig section, but that restricts the peer download to realtime only.
In the splitSegmentCommitter class, look at the table config to construct the URL.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418620328
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
+ }
+ return _defaultSegmentLocationURI;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS);
+ return segmentLocation;
+ } catch (Exception e) {
+ LOGGER.error("Failed to upload file {} of segment {} for table {} ", segmentFile.getAbsolutePath(), segmentName,
Review comment:
Switch to use WARN. Are you asking about 2 error (or warning) messages: one for the callable task and the other for the future waiting? I could not think of cases in which both error will be logged. Even in the timeout scenario, only the future waiting error is logged -- as the unit test shows. Also the two errors have different signatures.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418733551
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
Review comment:
We can override the split commit class with another class, or pass a flag into the split committer to decide whether to override failures with default. I would prefer the former. The decision of which committer to use should be left to the table config.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418222957
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
+ }
+ return _defaultSegmentLocationURI;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS);
+ return segmentLocation;
+ } catch (Exception e) {
Review comment:
InterruptedException should be OK, if the service is being shutdown
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
Review comment:
```suggestion
public BestEffortSegmentUploader(String segmentStoreDirUri, int timeoutMillis, String defaultSegment) {
```
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
Review comment:
```suggestion
public class PinotFSSegmentUploader implements SegmentUploader {
```
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
Review comment:
can u javadoc the arguments? What is `defaultSegment`? Maybe call it defaultSegmentURI and define it as URI type?
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
Review comment:
Not sure if this should be logged as an error, since we have recovery mechanisms. Warning should be ok. We should log an error in higher layers if we just cannot recover from this
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
Review comment:
Better to separate the default uri functionality. If the uploader.upload fails, it always returns null (have a consistent javaoc in `SegmentUploader` interface). The caller can decide whether default is needed. In this case, the split committer should decide that.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/Server2ControllerSegmentUploader.java
##########
@@ -52,7 +52,7 @@ public Server2ControllerSegmentUploader(Logger segmentLogger, FileUploadDownload
}
@Override
- public URI uploadSegment(File segmentFile) {
+ public URI uploadSegment(File segmentFile, String tableName, String segmentName) {
Review comment:
```suggestion
public URI uploadSegment(File segmentFile, LLCSegmentName segmentName) {
```
`segmentName` already has the (raw) tableName.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
+ }
+ return _defaultSegmentLocationURI;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS);
+ return segmentLocation;
+ } catch (Exception e) {
+ LOGGER.error("Failed to upload file {} of segment {} for table {} ", segmentFile.getAbsolutePath(), segmentName,
Review comment:
same comment here about error. Will we be logging 2 error messages?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418809103
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418614604
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
Review comment:
parameter removed with new class constructor.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418615923
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
Review comment:
done. Another option is to use a AlwaysSucceedSegmentUploader with a default uri which wraps around the PinotFSSegmentUploader. For now, just let split committer handle the null case as you suggested.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418787425
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java
##########
@@ -0,0 +1,84 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr) using PinotFS within a configurable timeout period. The final segment location would be in the
+// URI _segmentStoreUriStr/_tableNameWithType/segmentName if successful.
+public class PinotFSSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(PinotFSSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ private int _timeoutInMs;
+
+ public PinotFSSegmentUploader(String segmentStoreDirUri, int timeoutMillis) {
+ _segmentStoreUriStr = segmentStoreDirUri;
+ _timeoutInMs = timeoutMillis;
+ }
+
+ public URI uploadSegment(File segmentFile, LLCSegmentName segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return null;
+ }
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil
+ .join(File.separator, _segmentStoreUriStr, segmentName.getTableName(), segmentName.getSegmentName()));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.warn("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r419561887
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactory.java
##########
@@ -29,18 +29,20 @@
*/
public class SegmentCommitterFactory {
private static Logger LOGGER;
- private final IndexLoadingConfig _indexLoadingConfig;
+ private final TableConfig _tableConfig;
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
- public SegmentCommitterFactory(Logger segmentLogger, IndexLoadingConfig indexLoadingConfig,
+ public SegmentCommitterFactory(Logger segmentLogger, TableConfig tableConfig,
Review comment:
My bad, looks like tableConfig is not used, we can remove it.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java
##########
@@ -0,0 +1,86 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A segment uploader which does segment upload to a segment store (with store root dir configured as
+ * _segmentStoreUriStr) using PinotFS within a configurable timeout period. The final segment location would be in the
+ * URI _segmentStoreUriStr/_tableNameWithType/segmentName if successful.
+ */
+public class PinotFSSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(PinotFSSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ private int _timeoutInMs;
+
+ public PinotFSSegmentUploader(String segmentStoreDirUri, int timeoutMillis) {
+ _segmentStoreUriStr = segmentStoreDirUri;
+ _timeoutInMs = timeoutMillis;
+ }
+
+ public URI uploadSegment(File segmentFile, LLCSegmentName segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return null;
+ }
+ Callable<URI> uploadTask = () -> {
+ URI destUri = new URI(StringUtil
+ .join(File.separator, _segmentStoreUriStr, segmentName.getTableName(), segmentName.getSegmentName()));
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.warn("Failed copy segment tar file {} to segment store {}: {}", segmentFile.getName(), destUri, e);
+ }
+ return null;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS);
+ return segmentLocation;
+ } catch (InterruptedException e) {
+ LOGGER.info("Interrupted while waiting for segment upload.");
Review comment:
```suggestion
LOGGER.info("Interrupted while waiting for segment upload of {} to {}.", segmentName, _segmentStoreUriStr);
```
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/Server2ControllerSegmentUploader.java
##########
@@ -24,6 +24,7 @@
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.common.utils.LLCSegmentName;
Review comment:
In this class, we should not need the segment name argument anymore since we get it in the upload() call.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418808964
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java
##########
@@ -35,16 +38,34 @@
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
private final IndexLoadingConfig _indexLoadingConfig;
private final SegmentUploader _segmentUploader;
+ // The default segment location uri str, could be null.
+ private final String _defaultSegmentLocation;
private final Logger _segmentLogger;
public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader) {
+ this(segmentLogger, protocolHandler, indexLoadingConfig, params, segmentUploader, null);
+ }
+
+ /**
+ *
+ * @param segmentLogger
+ * @param protocolHandler
+ * @param indexLoadingConfig
+ * @param params
+ * @param segmentUploader
+ * @param defaultSegmentLocation The default segment location uri str, could be null.
+ */
+ public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
+ IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader,
+ String defaultSegmentLocation) {
Review comment:
Will send out the config change proposal. Replace indexLoadingConfig with tableConfig. Note that isEnableSplitCommitEndWithMetadata in indexLoadingConfig is a server instance level config. TableConfig does not have this field -- so I add a new boolean flag to the factory creation method. This flag can be removed if no other is using the deprecated _protocolHandler.segmentCommitEnd() anymore.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r419558050
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java
##########
@@ -35,16 +38,34 @@
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
private final IndexLoadingConfig _indexLoadingConfig;
private final SegmentUploader _segmentUploader;
+ // The default segment location uri str, could be null.
+ private final String _defaultSegmentLocation;
private final Logger _segmentLogger;
public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader) {
+ this(segmentLogger, protocolHandler, indexLoadingConfig, params, segmentUploader, null);
+ }
+
+ /**
+ *
+ * @param segmentLogger
+ * @param protocolHandler
+ * @param indexLoadingConfig
+ * @param params
+ * @param segmentUploader
+ * @param defaultSegmentLocation The default segment location uri str, could be null.
+ */
+ public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
+ IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader,
+ String defaultSegmentLocation) {
Review comment:
Ah, the config isEnableSplitCommitEndWithMetadata can re removed now. We had introduced that for upgrade compatibility. by now, the controllers should already have the capability to get split commit with metadata.
So, you can remove that, and then, like you pointed out, remove tableConfig as well (for some reason, I thought tableConfig had that boolean. my bad, sorry)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io commented on pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#issuecomment-622285696
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=h1) Report
> Merging [#5314](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/bd79861e7bca87406fc184ec70ce504667f76daf&el=desc) will **decrease** coverage by `0.16%`.
> The diff coverage is `72.41%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #5314 +/- ##
==========================================
- Coverage 66.23% 66.07% -0.17%
==========================================
Files 1066 1073 +7
Lines 54552 54595 +43
Branches 8128 8141 +13
==========================================
- Hits 36132 36072 -60
- Misses 15774 15884 +110
+ Partials 2646 2639 -7
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...e/pinot/broker/api/resources/PinotBrokerDebug.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYXBpL3Jlc291cmNlcy9QaW5vdEJyb2tlckRlYnVnLmphdmE=) | `76.66% <ø> (ø)` | |
| [.../BrokerResourceOnlineOfflineStateModelFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclJlc291cmNlT25saW5lT2ZmbGluZVN0YXRlTW9kZWxGYWN0b3J5LmphdmE=) | `55.81% <ø> (ø)` | |
| [...quota/HelixExternalViewBasedQueryQuotaManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IZWxpeEV4dGVybmFsVmlld0Jhc2VkUXVlcnlRdW90YU1hbmFnZXIuamF2YQ==) | `67.87% <ø> (ø)` | |
| [...roker/requesthandler/BaseBrokerRequestHandler.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcmVxdWVzdGhhbmRsZXIvQmFzZUJyb2tlclJlcXVlc3RIYW5kbGVyLmphdmE=) | `79.35% <ø> (ø)` | |
| [...rg/apache/pinot/broker/routing/RoutingManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9Sb3V0aW5nTWFuYWdlci5qYXZh) | `81.92% <ø> (+0.76%)` | :arrow_up: |
| [...ting/instanceselector/InstanceSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3JGYWN0b3J5LmphdmE=) | `71.42% <ø> (ø)` | |
| [...er/routing/segmentpruner/SegmentPrunerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1NlZ21lbnRQcnVuZXJGYWN0b3J5LmphdmE=) | `83.33% <ø> (ø)` | |
| [...outing/segmentselector/SegmentSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50c2VsZWN0b3IvU2VnbWVudFNlbGVjdG9yRmFjdG9yeS5qYXZh) | `60.00% <ø> (ø)` | |
| [...mmon/assignment/InstanceAssignmentConfigUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vYXNzaWdubWVudC9JbnN0YW5jZUFzc2lnbm1lbnRDb25maWdVdGlscy5qYXZh) | `67.50% <ø> (-5.00%)` | :arrow_down: |
| [...not/common/assignment/InstancePartitionsUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vYXNzaWdubWVudC9JbnN0YW5jZVBhcnRpdGlvbnNVdGlscy5qYXZh) | `73.17% <ø> (ø)` | |
| ... and [231 more](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=footer). Last update [1fe22b5...4f5e855](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r419714696
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/Server2ControllerSegmentUploader.java
##########
@@ -24,6 +24,7 @@
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.common.utils.LLCSegmentName;
Review comment:
This is a compile time dependency because this uploader implements the interface's
URI uploadSegment(File segmentFile, LLCSegmentName segmentName);
method. The interface method uses segmentName -- which is a reasonable choice given that some implementation may need that info e.g., PinotFSUploader.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#issuecomment-622578544
> You can just change the arguments of SplitSegmentCommitter constructor to include TableConfig instea of indexLoadingConfig and commit this PR without actually handling the null case. After we finalize the config, we can handle the null case.
>
> Meanwhile, we can proceed with other PRs (e.g. a peer downloader) towards finishing the feature
I will file the peer download PR soon. We can do these two PRs in parallel. @mcvsubbu
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r418616022
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
+ }
+ return _defaultSegmentLocationURI;
+ };
+ Future<URI> future = _executorService.submit(uploadTask);
+ try {
+ URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS);
+ return segmentLocation;
+ } catch (Exception e) {
Review comment:
done.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
Review comment:
use warning.
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/BestEffortSegmentUploader.java
##########
@@ -0,0 +1,85 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import java.io.File;
+import java.net.URI;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.utils.StringUtil;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+// A segment uploader which does best effort segment upload to a segment store (with store root dir configured as
+// _segmentStoreUriStr). The upload is successful if it is done within a configurable timeout period.
+// The final segment location would be in the URI _segmentStoreUriStr/_tableNameWithType/segmentName if
+// successful. If a segment upload fails or there is no segment store uri configured, it sets the segment location as
+// the default URI.
+public class BestEffortSegmentUploader implements SegmentUploader {
+ private Logger LOGGER = LoggerFactory.getLogger(BestEffortSegmentUploader.class);
+ private String _segmentStoreUriStr;
+ private ExecutorService _executorService = Executors.newCachedThreadPool();
+ // The default segment location URI to return if the upload fails.
+ private URI _defaultSegmentLocationURI;
+ private int _timeoutInMs;
+
+ public BestEffortSegmentUploader(String segmentStoreUriStr, int timeoutInMs, String defaultSegment) {
+ _segmentStoreUriStr = segmentStoreUriStr;
+ _timeoutInMs = timeoutInMs;
+ _defaultSegmentLocationURI = URI.create(defaultSegment);
+ }
+
+ public URI uploadSegment(File segmentFile, String tableNameWithType, String segmentName) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ return _defaultSegmentLocationURI;
+ }
+
+ Callable<URI> uploadTask = () -> {
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme());
+ URI destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr, tableNameWithType, segmentName));
+ // Check and delete any existing segment file.
+ if (pinotFS.exists(destUri)) {
+ pinotFS.delete(destUri, true);
+ }
+ pinotFS.copyFromLocalFile(segmentFile, destUri);
+ return destUri;
+ } catch (Exception e) {
+ LOGGER.error("Failed copy segment tar file to segment store {}: {}", segmentFile.getName(), e);
Review comment:
Done and use warning.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#issuecomment-622285696
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=h1) Report
> Merging [#5314](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/bd79861e7bca87406fc184ec70ce504667f76daf&el=desc) will **decrease** coverage by `0.12%`.
> The diff coverage is `72.61%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #5314 +/- ##
==========================================
- Coverage 66.23% 66.10% -0.13%
==========================================
Files 1066 1073 +7
Lines 54552 54584 +32
Branches 8128 8138 +10
==========================================
- Hits 36132 36083 -49
- Misses 15774 15862 +88
+ Partials 2646 2639 -7
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...e/pinot/broker/api/resources/PinotBrokerDebug.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYXBpL3Jlc291cmNlcy9QaW5vdEJyb2tlckRlYnVnLmphdmE=) | `76.66% <ø> (ø)` | |
| [.../BrokerResourceOnlineOfflineStateModelFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclJlc291cmNlT25saW5lT2ZmbGluZVN0YXRlTW9kZWxGYWN0b3J5LmphdmE=) | `55.81% <ø> (ø)` | |
| [...quota/HelixExternalViewBasedQueryQuotaManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IZWxpeEV4dGVybmFsVmlld0Jhc2VkUXVlcnlRdW90YU1hbmFnZXIuamF2YQ==) | `67.87% <ø> (ø)` | |
| [...roker/requesthandler/BaseBrokerRequestHandler.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcmVxdWVzdGhhbmRsZXIvQmFzZUJyb2tlclJlcXVlc3RIYW5kbGVyLmphdmE=) | `79.35% <ø> (ø)` | |
| [...rg/apache/pinot/broker/routing/RoutingManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9Sb3V0aW5nTWFuYWdlci5qYXZh) | `81.92% <ø> (+0.76%)` | :arrow_up: |
| [...ting/instanceselector/InstanceSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3JGYWN0b3J5LmphdmE=) | `71.42% <ø> (ø)` | |
| [...er/routing/segmentpruner/SegmentPrunerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1NlZ21lbnRQcnVuZXJGYWN0b3J5LmphdmE=) | `83.33% <ø> (ø)` | |
| [...outing/segmentselector/SegmentSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50c2VsZWN0b3IvU2VnbWVudFNlbGVjdG9yRmFjdG9yeS5qYXZh) | `60.00% <ø> (ø)` | |
| [...oker/routing/timeboundary/TimeBoundaryManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy90aW1lYm91bmRhcnkvVGltZUJvdW5kYXJ5TWFuYWdlci5qYXZh) | `89.74% <ø> (+2.24%)` | :arrow_up: |
| [...mmon/assignment/InstanceAssignmentConfigUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vYXNzaWdubWVudC9JbnN0YW5jZUFzc2lnbm1lbnRDb25maWdVdGlscy5qYXZh) | `72.50% <ø> (ø)` | |
| ... and [232 more](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=footer). Last update [1fe22b5...2d53c0d](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r419712623
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java
##########
@@ -35,16 +38,34 @@
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
private final IndexLoadingConfig _indexLoadingConfig;
private final SegmentUploader _segmentUploader;
+ // The default segment location uri str, could be null.
+ private final String _defaultSegmentLocation;
private final Logger _segmentLogger;
public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader) {
+ this(segmentLogger, protocolHandler, indexLoadingConfig, params, segmentUploader, null);
+ }
+
+ /**
+ *
+ * @param segmentLogger
+ * @param protocolHandler
+ * @param indexLoadingConfig
+ * @param params
+ * @param segmentUploader
+ * @param defaultSegmentLocation The default segment location uri str, could be null.
+ */
+ public SplitSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
+ IndexLoadingConfig indexLoadingConfig, SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader,
+ String defaultSegmentLocation) {
Review comment:
the boolean flag isEnableSplitCommitEndWithMetadata and tableConfig removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-io edited a comment on pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#issuecomment-622285696
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=h1) Report
> Merging [#5314](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/bd79861e7bca87406fc184ec70ce504667f76daf&el=desc) will **decrease** coverage by `0.12%`.
> The diff coverage is `72.61%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #5314 +/- ##
==========================================
- Coverage 66.23% 66.10% -0.13%
==========================================
Files 1066 1073 +7
Lines 54552 54584 +32
Branches 8128 8138 +10
==========================================
- Hits 36132 36083 -49
- Misses 15774 15862 +88
+ Partials 2646 2639 -7
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...e/pinot/broker/api/resources/PinotBrokerDebug.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYXBpL3Jlc291cmNlcy9QaW5vdEJyb2tlckRlYnVnLmphdmE=) | `76.66% <ø> (ø)` | |
| [.../BrokerResourceOnlineOfflineStateModelFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL2hlbGl4L0Jyb2tlclJlc291cmNlT25saW5lT2ZmbGluZVN0YXRlTW9kZWxGYWN0b3J5LmphdmE=) | `55.81% <ø> (ø)` | |
| [...quota/HelixExternalViewBasedQueryQuotaManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcXVlcnlxdW90YS9IZWxpeEV4dGVybmFsVmlld0Jhc2VkUXVlcnlRdW90YU1hbmFnZXIuamF2YQ==) | `67.87% <ø> (ø)` | |
| [...roker/requesthandler/BaseBrokerRequestHandler.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcmVxdWVzdGhhbmRsZXIvQmFzZUJyb2tlclJlcXVlc3RIYW5kbGVyLmphdmE=) | `79.35% <ø> (ø)` | |
| [...rg/apache/pinot/broker/routing/RoutingManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9Sb3V0aW5nTWFuYWdlci5qYXZh) | `81.92% <ø> (+0.76%)` | :arrow_up: |
| [...ting/instanceselector/InstanceSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3JGYWN0b3J5LmphdmE=) | `71.42% <ø> (ø)` | |
| [...er/routing/segmentpruner/SegmentPrunerFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50cHJ1bmVyL1NlZ21lbnRQcnVuZXJGYWN0b3J5LmphdmE=) | `83.33% <ø> (ø)` | |
| [...outing/segmentselector/SegmentSelectorFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9zZWdtZW50c2VsZWN0b3IvU2VnbWVudFNlbGVjdG9yRmFjdG9yeS5qYXZh) | `60.00% <ø> (ø)` | |
| [...oker/routing/timeboundary/TimeBoundaryManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy90aW1lYm91bmRhcnkvVGltZUJvdW5kYXJ5TWFuYWdlci5qYXZh) | `89.74% <ø> (+2.24%)` | :arrow_up: |
| [...mmon/assignment/InstanceAssignmentConfigUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vYXNzaWdubWVudC9JbnN0YW5jZUFzc2lnbm1lbnRDb25maWdVdGlscy5qYXZh) | `72.50% <ø> (ø)` | |
| ... and [232 more](https://codecov.io/gh/apache/incubator-pinot/pull/5314/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=footer). Last update [1fe22b5...2d53c0d](https://codecov.io/gh/apache/incubator-pinot/pull/5314?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] chenboat commented on a change in pull request #5314: [Part 3.1] Deepstore by-pass: Add a new best effort segment uploader with bounded upload time and d…
Posted by GitBox <gi...@apache.org>.
chenboat commented on a change in pull request #5314:
URL: https://github.com/apache/incubator-pinot/pull/5314#discussion_r419716052
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactory.java
##########
@@ -29,18 +29,20 @@
*/
public class SegmentCommitterFactory {
private static Logger LOGGER;
- private final IndexLoadingConfig _indexLoadingConfig;
+ private final TableConfig _tableConfig;
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
- public SegmentCommitterFactory(Logger segmentLogger, IndexLoadingConfig indexLoadingConfig,
+ public SegmentCommitterFactory(Logger segmentLogger, TableConfig tableConfig,
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org