xiangfu0 commented on code in PR #19817:
URL: https://github.com/apache/pinot/pull/19817#discussion_r4290443898
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1765,26 +1766,46 @@ public void goOnlineFromConsuming(SegmentZKMetadata
segmentZKMetadata)
protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
throws Exception {
- boolean allowedDuringDownload =
_parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
// A local build that failed or mismatched the committed CRC has already
released the semaphore when the policy
// allows consumption during build, so the next consuming segment can be
running during this download as well
boolean releasedDuringBuild =
_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!_consumerSemaphoreAcquired.get();
- if ((allowedDuringDownload || releasedDuringBuild) &&
_partitionUpsertMetadataManager != null
- &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment))
{
- // Table config validation rejects this in PROTECTED mode. It still
happens when the mode is switched after the
- // table is created, when the server-level default allows partial upsert
consumption during commit, or when a
- // pauseless table falls back to a download after its local build fails.
- _segmentLogger.error("Next consuming segment can run during download
while this replace reverts upsert "
- + "metadata. Its snapshot can miss the rows the revert restores. Use
DISALLOW_ALWAYS");
- _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ // Hold the semaphore when the replace rewrites the key map, so the next
consuming segment does not start against
+ // a partially reverted one, and does not replay rows against keys this
replace is about to remove.
+ boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata();
+ boolean allowedDuringDownload =
+ _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() &&
!holdsConsumerSemaphore;
+ if (holdsConsumerSemaphore && (releasedDuringBuild
+ || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload())) {
+ _segmentLogger.warn("Policy {} would let the next consuming segment run
while this replace rewrites upsert or "
+ + "dedup metadata. Use DISALLOW_ALWAYS",
_parallelSegmentConsumptionPolicy);
+ // Only when the semaphore was already released: the rows are actually
at risk then.
+ if (releasedDuringBuild && revertsUpsertMetadata()) {
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
Review Comment:
This metric now looks unreachable. `releasedDuringBuild` requires the build
to have released, but `shouldReleaseSemaphoreBeforeBuild()` returns false
whenever `revertsUpsertMetadata()` is true, so the two guards here are mutually
exclusive (barring the revert flag flipping between build and download).
If the intent is to retire this signal, worth saying so; otherwise the only
case it was meant to catch can no longer fire.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1765,26 +1766,46 @@ public void goOnlineFromConsuming(SegmentZKMetadata
segmentZKMetadata)
protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
throws Exception {
- boolean allowedDuringDownload =
_parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
// A local build that failed or mismatched the committed CRC has already
released the semaphore when the policy
// allows consumption during build, so the next consuming segment can be
running during this download as well
boolean releasedDuringBuild =
_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!_consumerSemaphoreAcquired.get();
- if ((allowedDuringDownload || releasedDuringBuild) &&
_partitionUpsertMetadataManager != null
- &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment))
{
- // Table config validation rejects this in PROTECTED mode. It still
happens when the mode is switched after the
- // table is created, when the server-level default allows partial upsert
consumption during commit, or when a
- // pauseless table falls back to a download after its local build fails.
- _segmentLogger.error("Next consuming segment can run during download
while this replace reverts upsert "
- + "metadata. Its snapshot can miss the rows the revert restores. Use
DISALLOW_ALWAYS");
- _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ // Hold the semaphore when the replace rewrites the key map, so the next
consuming segment does not start against
+ // a partially reverted one, and does not replay rows against keys this
replace is about to remove.
+ boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata();
+ boolean allowedDuringDownload =
+ _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() &&
!holdsConsumerSemaphore;
+ if (holdsConsumerSemaphore && (releasedDuringBuild
+ || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload())) {
+ _segmentLogger.warn("Policy {} would let the next consuming segment run
while this replace rewrites upsert or "
+ + "dedup metadata. Use DISALLOW_ALWAYS",
_parallelSegmentConsumptionPolicy);
+ // Only when the semaphore was already released: the rows are actually
at risk then.
+ if (releasedDuringBuild && revertsUpsertMetadata()) {
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ }
}
if (allowedDuringDownload) {
closeStreamConsumerAndReleaseSemaphore();
}
_realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata);
}
+ /// Upsert reverts the keys at the replace; dedup re-points them to the new
immutable segment.
+ /// Extracted to be testable: the test double replaces
[#buildSegmentInternal] wholesale.
+ @VisibleForTesting
+ boolean shouldReleaseSemaphoreBeforeBuild() {
+ return _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!revertsUpsertMetadata();
+ }
+
+ private boolean holdsConsumerSemaphoreForMetadata() {
+ return revertsUpsertMetadata() || _partitionDedupMetadataManager != null;
Review Comment:
This claims to hold for dedup, but dedup only actually holds under
`ALLOW_DURING_DOWNLOAD_ONLY`. Under the pauseless default
(`ALLOW_DURING_BUILD_ONLY`), `shouldReleaseSemaphoreBeforeBuild()` is true for
dedup (revert flag is false), so the build already released and this hold is a
no-op.
So the dedup protection applies to the download-only policy but not the more
common build-only one. If that asymmetry is intended, a one-line note here
would help the next reader; otherwise dedup may still be exposed on the
build-release path.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
