This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new edffe1a832 Fix race condition when 2 segment upload occurred for the
same segment (#9905)
edffe1a832 is described below
commit edffe1a832b7a534ebc40d0cd1ca1bbccc1eb02e
Author: Jialiang Li <[email protected]>
AuthorDate: Wed Dec 7 10:05:12 2022 -0800
Fix race condition when 2 segment upload occurred for the same segment
(#9905)
Co-authored-by: Jack Li(Analytics Engineering) <[email protected]>
---
.../pinot/controller/api/upload/ZKOperator.java | 49 +++++++++++++++++-----
.../helix/core/SegmentDeletionManager.java | 1 -
2 files changed, 39 insertions(+), 11 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
index 3d33a358f3..4229d6c433 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java
@@ -297,8 +297,9 @@ public class ZKOperator {
}
// Lock if enableParallelPushProtection is true.
+ long segmentUploadStartTime = System.currentTimeMillis();
if (enableParallelPushProtection) {
-
newSegmentZKMetadata.setSegmentUploadStartTime(System.currentTimeMillis());
+ newSegmentZKMetadata.setSegmentUploadStartTime(segmentUploadStartTime);
}
// Update zk metadata customer map
@@ -323,8 +324,7 @@ public class ZKOperator {
// Cleanup the Zk entry and the segment from the permanent directory
if it exists.
LOGGER.error("Could not move segment {} from table {} to permanent
directory", segmentName, tableNameWithType,
e);
- _pinotHelixResourceManager.deleteSegment(tableNameWithType,
segmentName);
- LOGGER.info("Deleted zk entry and segment {} for table {}.",
segmentName, tableNameWithType);
+ deleteSegmentIfNeeded(tableNameWithType, segmentName,
segmentUploadStartTime, enableParallelPushProtection);
throw e;
}
}
@@ -332,11 +332,11 @@ public class ZKOperator {
try {
_pinotHelixResourceManager.assignTableSegment(tableNameWithType,
segmentMetadata.getName());
} catch (Exception e) {
- // assignTableSegment removes the zk entry. Call deleteSegment to remove
the segment from permanent location.
+ // assignTableSegment removes the zk entry.
+ // Call deleteSegment to remove the segment from permanent location if
needed.
LOGGER.error("Caught exception while calling assignTableSegment for
adding segment: {} to table: {}", segmentName,
tableNameWithType, e);
- _pinotHelixResourceManager.deleteSegment(tableNameWithType, segmentName);
- LOGGER.info("Deleted zk entry and segment {} for table {}.",
segmentName, tableNameWithType);
+ deleteSegmentIfNeeded(tableNameWithType, segmentName,
segmentUploadStartTime, enableParallelPushProtection);
throw e;
}
@@ -344,14 +344,43 @@ public class ZKOperator {
// Release lock. Expected version will be 0 as we hold a lock and no
updates could take place meanwhile.
newSegmentZKMetadata.setSegmentUploadStartTime(-1);
if (!_pinotHelixResourceManager.updateZkMetadata(tableNameWithType,
newSegmentZKMetadata, 0)) {
- _pinotHelixResourceManager.deleteSegment(tableNameWithType,
segmentName);
- LOGGER.info("Deleted zk entry and segment {} for table {}.",
segmentName, tableNameWithType);
- throw new RuntimeException(
- String.format("Failed to update ZK metadata for segment: %s of
table: %s", segmentFile, tableNameWithType));
+ // There is a race condition when it took too much time for the 1st
segment upload to process (due to slow
+ // PinotFS access), which leads to the 2nd attempt of segment upload,
and the 2nd segment upload succeeded.
+ // In this case, when the 1st upload comes back, it shouldn't blindly
delete the segment when it failed to
+ // update the zk metadata. Instead, the 1st attempt should validate
the upload start time one more time. If the
+ // start time doesn't match with the one persisted in zk metadata,
segment deletion should be skipped.
+ String errorMsg =
+ String.format("Failed to update ZK metadata for segment: %s of
table: %s", segmentFile, tableNameWithType);
+ LOGGER.error(errorMsg);
+ deleteSegmentIfNeeded(tableNameWithType, segmentName,
segmentUploadStartTime, true);
+ throw new RuntimeException(errorMsg);
}
}
}
+ /**
+ * Deletes the segment to be uploaded if either one of the criteria is
qualified:
+ * 1) the uploadStartTime matches with the one persisted in ZK metadata.
+ * 2) enableParallelPushProtection is not enabled.
+ */
+ private void deleteSegmentIfNeeded(String tableNameWithType, String
segmentName, long currentSegmentUploadStartTime,
+ boolean enableParallelPushProtection) {
+ ZNRecord existingSegmentMetadataZNRecord =
+
_pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType,
segmentName);
+ if (existingSegmentMetadataZNRecord == null) {
+ return;
+ }
+ // Check if the upload start time is set by this thread itself, if yes
delete the segment.
+ SegmentZKMetadata segmentZKMetadata = new
SegmentZKMetadata(existingSegmentMetadataZNRecord);
+ long existingSegmentUploadStartTime =
segmentZKMetadata.getSegmentUploadStartTime();
+ LOGGER.info("Parallel push protection is {} for segment: {}.",
+ (enableParallelPushProtection ? "enabled" : "disabled"), segmentName);
+ if (!enableParallelPushProtection || currentSegmentUploadStartTime ==
existingSegmentUploadStartTime) {
+ _pinotHelixResourceManager.deleteSegment(tableNameWithType, segmentName);
+ LOGGER.info("Deleted zk entry and segment {} for table {}.",
segmentName, tableNameWithType);
+ }
+ }
+
private void copySegmentToDeepStore(String tableNameWithType, String
segmentName, FileUploadType uploadType,
File segmentFile, String sourceDownloadURIStr, URI
finalSegmentLocationURI)
throws Exception {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
index 570a263ac8..ff74bea708 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
@@ -193,7 +193,6 @@ public class SegmentDeletionManager {
long effectiveDeletionDelay = Math.min(deletionDelay * 2,
MAX_DELETION_DELAY_SECONDS);
LOGGER.info("Postponing deletion of {} segments from table {}",
segmentsToRetryLater.size(), tableName);
deleteSegmentsWithDelay(tableName, segmentsToRetryLater,
deletedSegmentsRetentionMs, effectiveDeletionDelay);
- return;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
