You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2020/10/20 21:43:32 UTC
[incubator-pinot] branch add_table_lock_for_segment_upload created
(now f3a94cc)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a change to branch add_table_lock_for_segment_upload
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at f3a94cc Add table level lock for segment upload
This branch includes the following new commits:
new f3a94cc Add table level lock for segment upload
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: Add table level lock for segment upload
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch add_table_lock_for_segment_upload
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit f3a94cc976c4a63509f0844db14b0cf5c95e46fd
Author: Xiang Fu <fx...@gmail.com>
AuthorDate: Tue Oct 20 14:43:06 2020 -0700
Add table level lock for segment upload
---
.../helix/core/PinotHelixResourceManager.java | 41 +++++++++++++---------
1 file changed, 25 insertions(+), 16 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index ff519b9..bd15326 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -140,6 +140,8 @@ public class PinotHelixResourceManager {
private final Map<String, Map<String, Long>> _segmentCrcMap = new HashMap<>();
private final Map<String, Map<String, Integer>> _lastKnownSegmentMetadataVersionMap = new HashMap<>();
+ private final Map<String, Object> _tableUpdaterLock = new HashMap<>();
+
private final LoadingCache<String, String> _instanceAdminEndpointCache;
private final String _helixZkURL;
@@ -1641,23 +1643,30 @@ public class PinotHelixResourceManager {
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap = Collections
.singletonMap(InstancePartitionsType.OFFLINE, InstancePartitionsUtils
.fetchOrComputeInstancePartitions(_helixZkManager, offlineTableConfig, InstancePartitionsType.OFFLINE));
- HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
- assert idealState != null;
- Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
- if (currentAssignment.containsKey(segmentName)) {
- LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
- offlineTableName);
- } else {
- List<String> assignedInstances =
- segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
- LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
- offlineTableName);
- currentAssignment.put(segmentName,
- SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ if (_tableUpdaterLock.get(offlineTableName) == null) {
+ synchronized (_tableUpdaterLock) {
+ if (_tableUpdaterLock.get(offlineTableName) == null) {
+ _tableUpdaterLock.put(offlineTableName, new Object());
+ }
}
- return idealState;
- });
- LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ }
+ synchronized (_tableUpdaterLock.get(offlineTableName)) {
+ HelixHelper.updateIdealState(_helixZkManager, offlineTableName, idealState -> {
+ assert idealState != null;
+ Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
+ if (currentAssignment.containsKey(segmentName)) {
+ LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
+ offlineTableName);
+ } else {
+ List<String> assignedInstances = segmentAssignment.assignSegment(segmentName, currentAssignment, instancePartitionsMap);
+ LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
+ offlineTableName);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances, SegmentStateModel.ONLINE));
+ }
+ return idealState;
+ });
+ LOGGER.info("Added segment: {} to IdealState for table: {}", segmentName, offlineTableName);
+ }
} catch (Exception e) {
LOGGER
.error("Caught exception while adding segment: {} to IdealState for table: {}, deleting segment ZK metadata",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org