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 2024/09/07 15:48:38 UTC
(pinot) branch master updated: Acquire IdealStateUpdaterLock for PinotLLCRealtimeSegmentManager IdealState update (#13947)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new f5327a512a Acquire IdealStateUpdaterLock for PinotLLCRealtimeSegmentManager IdealState update (#13947)
f5327a512a is described below
commit f5327a512ab90b6137c88c2fbaf0c517c8ac50b6
Author: Xiang Fu <xi...@gmail.com>
AuthorDate: Sat Sep 7 08:48:32 2024 -0700
Acquire IdealStateUpdaterLock for PinotLLCRealtimeSegmentManager IdealState update (#13947)
---
.../helix/core/PinotHelixResourceManager.java | 88 ++++++++--------
.../realtime/PinotLLCRealtimeSegmentManager.java | 111 +++++++++++----------
2 files changed, 104 insertions(+), 95 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 f3a4819c6f..f727ea5dbc 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
@@ -1980,11 +1980,13 @@ public class PinotHelixResourceManager {
IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableNameWithType);
String replicationConfigured = Integer.toString(tableConfig.getReplication());
if (!idealState.getReplicas().equals(replicationConfigured)) {
- HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, is -> {
- assert is != null;
- is.setReplicas(replicationConfigured);
- return is;
- }, RetryPolicies.exponentialBackoffRetryPolicy(5, 1000L, 1.2f));
+ synchronized (getIdealStateUpdaterLock(tableNameWithType)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, is -> {
+ assert is != null;
+ is.setReplicas(replicationConfigured);
+ return is;
+ }, RetryPolicies.exponentialBackoffRetryPolicy(5, 1000L, 1.2f));
+ }
}
// Assign instances
@@ -3967,48 +3969,48 @@ public class PinotHelixResourceManager {
LineageEntry lineageEntryToUpdate, LineageEntry lineageEntryToMatch, ZkHelixPropertyStore<ZNRecord> propertyStore,
LineageUpdateType lineageUpdateType, Map<String, String> customMap) {
String tableNameWithType = tableConfig.getTableName();
- synchronized (getLineageUpdaterLock(tableNameWithType)) {
- // retry attempts are made to account for the distributed update from other controllers
- for (int i = 0; i < DEFAULT_SEGMENT_LINEAGE_UPDATE_NUM_RETRY; i++) {
- // Fetch the segment lineage
- ZNRecord segmentLineageToUpdateZNRecord =
- SegmentLineageAccessHelper.getSegmentLineageZNRecord(propertyStore, tableConfig.getTableName());
- int expectedVersion = segmentLineageToUpdateZNRecord.getVersion();
- SegmentLineage segmentLineageToUpdate = SegmentLineage.fromZNRecord(segmentLineageToUpdateZNRecord);
- LineageEntry currentLineageEntry = segmentLineageToUpdate.getLineageEntry(lineageEntryId);
-
- // If the lineage entry doesn't match with the previously fetched lineage, we need to fail the request.
- if (!currentLineageEntry.equals(lineageEntryToMatch)) {
- String errorMsg = String.format(
- "Aborting the to update lineage entry since we find that the entry has been modified for table %s, "
- + "entry id: %s", tableConfig.getTableName(), lineageEntryId);
- LOGGER.error(errorMsg);
- throw new RuntimeException(errorMsg);
- }
+ synchronized (getLineageUpdaterLock(tableNameWithType)) {
+ // retry attempts are made to account for the distributed update from other controllers
+ for (int i = 0; i < DEFAULT_SEGMENT_LINEAGE_UPDATE_NUM_RETRY; i++) {
+ // Fetch the segment lineage
+ ZNRecord segmentLineageToUpdateZNRecord =
+ SegmentLineageAccessHelper.getSegmentLineageZNRecord(propertyStore, tableConfig.getTableName());
+ int expectedVersion = segmentLineageToUpdateZNRecord.getVersion();
+ SegmentLineage segmentLineageToUpdate = SegmentLineage.fromZNRecord(segmentLineageToUpdateZNRecord);
+ LineageEntry currentLineageEntry = segmentLineageToUpdate.getLineageEntry(lineageEntryId);
+
+ // If the lineage entry doesn't match with the previously fetched lineage, we need to fail the request.
+ if (!currentLineageEntry.equals(lineageEntryToMatch)) {
+ String errorMsg = String.format(
+ "Aborting the to update lineage entry since we find that the entry has been modified for table %s, "
+ + "entry id: %s", tableConfig.getTableName(), lineageEntryId);
+ LOGGER.error(errorMsg);
+ throw new RuntimeException(errorMsg);
+ }
- // Update lineage entry
- segmentLineageToUpdate.updateLineageEntry(lineageEntryId, lineageEntryToUpdate);
- switch (lineageUpdateType) {
- case END:
- _lineageManager.updateLineageForEndReplaceSegments(tableConfig, lineageEntryId, customMap,
- segmentLineageToUpdate);
- break;
- case REVERT:
- _lineageManager.updateLineageForRevertReplaceSegments(tableConfig, lineageEntryId, customMap,
- segmentLineageToUpdate);
- break;
- default:
- String errorMsg = String.format("Aborting the lineage entry update with type: %s, as the allowed update"
- + "types in this method are END and REVERT", lineageUpdateType);
- throw new IllegalStateException(errorMsg);
- }
+ // Update lineage entry
+ segmentLineageToUpdate.updateLineageEntry(lineageEntryId, lineageEntryToUpdate);
+ switch (lineageUpdateType) {
+ case END:
+ _lineageManager.updateLineageForEndReplaceSegments(tableConfig, lineageEntryId, customMap,
+ segmentLineageToUpdate);
+ break;
+ case REVERT:
+ _lineageManager.updateLineageForRevertReplaceSegments(tableConfig, lineageEntryId, customMap,
+ segmentLineageToUpdate);
+ break;
+ default:
+ String errorMsg = String.format("Aborting the lineage entry update with type: %s, as the allowed update"
+ + "types in this method are END and REVERT", lineageUpdateType);
+ throw new IllegalStateException(errorMsg);
+ }
- // Write back to the lineage entry
- if (SegmentLineageAccessHelper.writeSegmentLineage(propertyStore, segmentLineageToUpdate, expectedVersion)) {
- return true;
- }
+ // Write back to the lineage entry
+ if (SegmentLineageAccessHelper.writeSegmentLineage(propertyStore, segmentLineageToUpdate, expectedVersion)) {
+ return true;
}
}
+ }
return false;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index 85c5cad0de..910291ff00 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -816,23 +816,24 @@ public class PinotLLCRealtimeSegmentManager {
String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(llcSegmentName.getTableName());
String segmentName = llcSegmentName.getSegmentName();
LOGGER.info("Marking CONSUMING segment: {} OFFLINE on instance: {}", segmentName, instanceName);
-
- try {
- HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
- assert idealState != null;
- Map<String, String> stateMap = idealState.getInstanceStateMap(segmentName);
- String state = stateMap.get(instanceName);
- if (SegmentStateModel.CONSUMING.equals(state)) {
- stateMap.put(instanceName, SegmentStateModel.OFFLINE);
- } else {
- LOGGER.info("Segment {} in state {} when trying to register consumption stop from {}", segmentName, state,
- instanceName);
- }
- return idealState;
- }, RetryPolicies.exponentialBackoffRetryPolicy(10, 500L, 1.2f), true);
- } catch (Exception e) {
- _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_ZOOKEEPER_UPDATE_FAILURES, 1L);
- throw e;
+ synchronized (_helixResourceManager.getIdealStateUpdaterLock(realtimeTableName)) {
+ try {
+ HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
+ assert idealState != null;
+ Map<String, String> stateMap = idealState.getInstanceStateMap(segmentName);
+ String state = stateMap.get(instanceName);
+ if (SegmentStateModel.CONSUMING.equals(state)) {
+ stateMap.put(instanceName, SegmentStateModel.OFFLINE);
+ } else {
+ LOGGER.info("Segment {} in state {} when trying to register consumption stop from {}", segmentName, state,
+ instanceName);
+ }
+ return idealState;
+ }, RetryPolicies.exponentialBackoffRetryPolicy(10, 500L, 1.2f), true);
+ } catch (Exception e) {
+ _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_ZOOKEEPER_UPDATE_FAILURES, 1L);
+ throw e;
+ }
}
// We know that we have successfully set the idealstate to be OFFLINE.
// We can now do a best effort to reset the externalview to be OFFLINE if it is in ERROR state.
@@ -924,30 +925,33 @@ public class PinotLLCRealtimeSegmentManager {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
- HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
- assert idealState != null;
- boolean isTableEnabled = idealState.isEnabled();
- boolean isTablePaused = isTablePaused(idealState);
- boolean offsetsHaveToChange = offsetCriteria != null;
- if (isTableEnabled && !isTablePaused) {
- List<PartitionGroupConsumptionStatus> currentPartitionGroupConsumptionStatusList =
- offsetsHaveToChange
- ? Collections.emptyList() // offsets from metadata are not valid anymore; fetch for all partitions
- : getPartitionGroupConsumptionStatusList(idealState, streamConfig);
- OffsetCriteria originalOffsetCriteria = streamConfig.getOffsetCriteria();
- // Read the smallest offset when a new partition is detected
- streamConfig.setOffsetCriteria(offsetsHaveToChange ? offsetCriteria : OffsetCriteria.SMALLEST_OFFSET_CRITERIA);
- List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- getNewPartitionGroupMetadataList(streamConfig, currentPartitionGroupConsumptionStatusList);
- streamConfig.setOffsetCriteria(originalOffsetCriteria);
- return ensureAllPartitionsConsuming(tableConfig, streamConfig, idealState, newPartitionGroupMetadataList,
- recreateDeletedConsumingSegment, offsetCriteria);
- } else {
- LOGGER.info("Skipping LLC segments validation for table: {}, isTableEnabled: {}, isTablePaused: {}",
- realtimeTableName, isTableEnabled, isTablePaused);
- return idealState;
- }
- }, RetryPolicies.exponentialBackoffRetryPolicy(10, 1000L, 1.2f), true);
+ synchronized (_helixResourceManager.getIdealStateUpdaterLock(realtimeTableName)) {
+ HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
+ assert idealState != null;
+ boolean isTableEnabled = idealState.isEnabled();
+ boolean isTablePaused = isTablePaused(idealState);
+ boolean offsetsHaveToChange = offsetCriteria != null;
+ if (isTableEnabled && !isTablePaused) {
+ List<PartitionGroupConsumptionStatus> currentPartitionGroupConsumptionStatusList =
+ offsetsHaveToChange
+ ? Collections.emptyList() // offsets from metadata are not valid anymore; fetch for all partitions
+ : getPartitionGroupConsumptionStatusList(idealState, streamConfig);
+ OffsetCriteria originalOffsetCriteria = streamConfig.getOffsetCriteria();
+ // Read the smallest offset when a new partition is detected
+ streamConfig.setOffsetCriteria(
+ offsetsHaveToChange ? offsetCriteria : OffsetCriteria.SMALLEST_OFFSET_CRITERIA);
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList =
+ getNewPartitionGroupMetadataList(streamConfig, currentPartitionGroupConsumptionStatusList);
+ streamConfig.setOffsetCriteria(originalOffsetCriteria);
+ return ensureAllPartitionsConsuming(tableConfig, streamConfig, idealState, newPartitionGroupMetadataList,
+ recreateDeletedConsumingSegment, offsetCriteria);
+ } else {
+ LOGGER.info("Skipping LLC segments validation for table: {}, isTableEnabled: {}, isTablePaused: {}",
+ realtimeTableName, isTableEnabled, isTablePaused);
+ return idealState;
+ }
+ }, RetryPolicies.exponentialBackoffRetryPolicy(10, 1000L, 1.2f), true);
+ }
}
/**
@@ -995,7 +999,7 @@ public class PinotLLCRealtimeSegmentManager {
String pauseStateStr = idealState.getRecord().getSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE);
try {
if (pauseStateStr != null) {
- return JsonUtils.stringToObject(pauseStateStr, PauseState.class);
+ return JsonUtils.stringToObject(pauseStateStr, PauseState.class);
}
} catch (JsonProcessingException e) {
LOGGER.warn("Unable to parse the pause state from ideal state : {}", pauseStateStr);
@@ -1503,9 +1507,9 @@ public class PinotLLCRealtimeSegmentManager {
// Use this retention value to avoid the data racing between segment upload and retention management.
long retentionMs = TimeUnit.valueOf(validationConfig.getRetentionTimeUnit().toUpperCase())
- .toMillis(Long.parseLong(validationConfig.getRetentionTimeValue()));
+ .toMillis(Long.parseLong(validationConfig.getRetentionTimeValue()));
RetentionStrategy retentionStrategy = new TimeRetentionStrategy(TimeUnit.MILLISECONDS,
- retentionMs - MIN_TIME_BEFORE_SEGMENT_EXPIRATION_FOR_FIXING_DEEP_STORE_COPY_MILLIS);
+ retentionMs - MIN_TIME_BEFORE_SEGMENT_EXPIRATION_FOR_FIXING_DEEP_STORE_COPY_MILLIS);
PinotFS pinotFS = PinotFSFactory.create(URIUtils.getUri(_controllerConf.getDataDir()).getScheme());
@@ -1725,7 +1729,7 @@ public class PinotLLCRealtimeSegmentManager {
sendForceCommitMessageToServers(tableNameWithType, consumingSegments);
return new PauseStatusDetails(true, consumingSegments, reasonCode, comment != null ? comment
: "Pause flag is set. Consuming segments are being committed."
- + " Use /pauseStatus endpoint in a few moments to check if all consuming segments have been committed.",
+ + " Use /pauseStatus endpoint in a few moments to check if all consuming segments have been committed.",
new Timestamp(System.currentTimeMillis()).toString());
}
@@ -1756,13 +1760,16 @@ public class PinotLLCRealtimeSegmentManager {
PauseState.ReasonCode reasonCode, @Nullable String comment) {
PauseState pauseState = new PauseState(pause, reasonCode, comment,
new Timestamp(System.currentTimeMillis()).toString());
- IdealState updatedIdealState = HelixHelper.updateIdealState(_helixManager, tableNameWithType, idealState -> {
- ZNRecord znRecord = idealState.getRecord();
- znRecord.setSimpleField(PAUSE_STATE, pauseState.toJsonString());
- // maintain for backward compatibility
- znRecord.setSimpleField(IS_TABLE_PAUSED, Boolean.valueOf(pause).toString());
- return new IdealState(znRecord);
- }, RetryPolicies.noDelayRetryPolicy(3));
+ IdealState updatedIdealState;
+ synchronized (_helixResourceManager.getIdealStateUpdaterLock(tableNameWithType)) {
+ updatedIdealState = HelixHelper.updateIdealState(_helixManager, tableNameWithType, idealState -> {
+ ZNRecord znRecord = idealState.getRecord();
+ znRecord.setSimpleField(PAUSE_STATE, pauseState.toJsonString());
+ // maintain for backward compatibility
+ znRecord.setSimpleField(IS_TABLE_PAUSED, Boolean.valueOf(pause).toString());
+ return new IdealState(znRecord);
+ }, RetryPolicies.noDelayRetryPolicy(3));
+ }
LOGGER.info("Set 'pauseState' to {} in the Ideal State for table {}. "
+ "Also set 'isTablePaused' to {} for backward compatibility.", pauseState, tableNameWithType, pause);
return updatedIdealState;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org