This is an automated email from the ASF dual-hosted git repository.
KKcorps 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 678c8b07a90 Hold the consumer semaphore while a replace reverts upsert
metadata (#19817)
678c8b07a90 is described below
commit 678c8b07a90878662c2339d7701922b2cbb45ba5
Author: Chaitanya Deepthi <[email protected]>
AuthorDate: Thu Oct 22 20:09:02 2026 -0700
Hold the consumer semaphore while a replace reverts upsert metadata (#19817)
---
.../realtime/RealtimeSegmentDataManager.java | 57 ++++++++++++++++------
.../realtime/RealtimeSegmentDataManagerTest.java | 26 ++++++++--
2 files changed, 63 insertions(+), 20 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index c5209138ff4..df626b8e6c0 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -1207,7 +1207,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
@Nullable
protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit)
throws SegmentBuildFailureException {
- if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) {
+ if (shouldReleaseSemaphoreBeforeBuild()) {
closeStreamConsumerAndReleaseSemaphore();
}
// Do not allow building segment when table data manager is already shut
down
@@ -1765,26 +1765,51 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
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);
- }
- if (allowedDuringDownload) {
+ boolean holdsPermit = holdsConsumerSemaphoreForMetadata();
+ if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload() &&
!holdsPermit) {
closeStreamConsumerAndReleaseSemaphore();
}
+ // Under a policy that releases somewhere, the permit should still be
ours, so a missing one means the next
+ // consuming segment is already running. DISALLOW_ALWAYS never releases,
so a missing permit there just means this
+ // segment never consumed.
+ boolean policyReleasesSomewhere =
_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()
+ || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
+ if (holdsPermit && policyReleasesSomewhere &&
!_consumerSemaphoreAcquired.get()) {
+ _segmentLogger.error("Next consuming segment is running while this
replace rewrites upsert or dedup metadata, "
+ + "so its rows can be dropped against keys this replace removes.
Policy: {}. Use DISALLOW_ALWAYS",
+ _parallelSegmentConsumptionPolicy);
+ if (revertsUpsertMetadata()) {
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ }
+ }
_realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata);
}
+ /// A build that fails or mismatches the committed CRC falls back to a
download, and the permit cannot be taken back
+ /// then, so tables whose replace rewrites the key map keep it. Pauseless
dedup is the exception: it is the one case
+ /// where holding costs what pauseless exists to provide, and it skips the
CRC check, so only an outright build
+ /// failure leaves it exposed.
+ ///
+ /// Negation of [#holdsConsumerSemaphoreForMetadata()] plus that exception;
keep the two in step.
+ @VisibleForTesting
+ boolean shouldReleaseSemaphoreBeforeBuild() {
+ if (!_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) {
+ return false;
+ }
+ return !revertsUpsertMetadata()
+ && (_partitionDedupMetadataManager == null ||
PauselessConsumptionUtils.isPauselessEnabled(_tableConfig));
+ }
+
+ /// The replace re-points or reverts every key of the consuming segment, so
the next one must not run against it.
+ private boolean holdsConsumerSemaphoreForMetadata() {
+ return revertsUpsertMetadata() || _partitionDedupMetadataManager != null;
+ }
+
+ private boolean revertsUpsertMetadata() {
+ return _partitionUpsertMetadataManager != null
+ &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment);
+ }
+
protected long now() {
return System.currentTimeMillis();
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index 5ab454edf28..64f1b96efb9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -987,12 +987,29 @@ public class RealtimeSegmentDataManagerTest {
ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY,
serverMetrics)) {
Semaphore semaphore =
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore();
Assert.assertTrue(semaphore.tryAcquire());
+ // The policy would release, but the replace reverts upsert metadata, so
the guard keeps the permit.
segmentDataManager._releaseSemaphoreOnBuild = true;
+
Assert.assertFalse(segmentDataManager.shouldReleaseSemaphoreBeforeBuild());
runGoOnlineForCrcGuard(segmentDataManager, crcMetadata(finalOffsetValue,
12345L), false, finalOffsetValue);
Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled);
- Assert.assertEquals(semaphore.availablePermits(), 1);
- verify(serverMetrics).addMeteredTableValue(anyString(),
- eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
eq(1L));
+ Assert.assertEquals(semaphore.availablePermits(), 0);
+ verify(serverMetrics, never()).addMeteredTableValue(anyString(),
+ eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
anyLong());
+ }
+
+ // ALLOW_DURING_DOWNLOAD_ONLY: releasedDuringBuild is never true here, so
only the override path can report it.
+ _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID);
+ serverMetrics = spy(new
ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()));
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManagerForUpsertRevert(
+ ParallelSegmentConsumptionPolicy.ALLOW_DURING_DOWNLOAD_ONLY,
serverMetrics)) {
+ Semaphore semaphore =
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore();
+ Assert.assertTrue(semaphore.tryAcquire());
+ segmentDataManager.getConsumerSemaphoreAcquired().set(true);
+
Assert.assertFalse(segmentDataManager.shouldReleaseSemaphoreBeforeBuild());
+
segmentDataManager.downloadSegmentAndReplace(crcMetadata(finalOffsetValue,
12345L));
+ Assert.assertEquals(semaphore.availablePermits(), 0);
+ verify(serverMetrics, never()).addMeteredTableValue(anyString(),
+ eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD),
anyLong());
}
// ALLOW_DURING_BUILD_ONLY without a local build: the download still holds
the semaphore, so nothing is flagged.
@@ -1891,7 +1908,8 @@ public class RealtimeSegmentDataManagerTest {
protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) {
terminateLoopIfNecessary();
_buildSegmentCalled = true;
- if (_releaseSemaphoreOnBuild) {
+ // Through the real predicate, so the guard is exercised rather than
bypassed.
+ if (_releaseSemaphoreOnBuild && shouldReleaseSemaphoreBeforeBuild()) {
closeStreamConsumerAndReleaseSemaphore();
}
if (_failSegmentBuild) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
