xiangfu0 commented on code in PR #19770:
URL: https://github.com/apache/pinot/pull/19770#discussion_r4203464094
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,173 @@ protected void processTable(String tableNameWithType) {
// Delete segments based on segment lineage and clean up segment lineage
metadata.
manageSegmentLineageCleanupForTable(tableConfig);
+
+ // Re-read active segments after time and lineage cleanup so
already-deleted bytes do not cause extra eviction.
+ manageSizeBasedRetention(tableConfig);
+ }
+
+ /// Enforces the optional compressed-segment size limit on live data
independently of time retention. Replaced
+ /// segments are excluded from accounting, and an in-progress replacement
postpones the entire pass. The newest
+ /// OFFLINE segment and realtime recovery segments are preserved; hybrid
realtime eviction requires offline coverage.
+ /// Unknown sizes or missing metadata disable this pass; consuming segments
are excluded from accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String tableNameWithType = tableConfig.getTableName();
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null || (tableConfig.getTableType() ==
TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig))))
{
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
Review Comment:
Fixed in
[c32ac7fa3a](https://github.com/apache/pinot/commit/c32ac7fa3aeb8cc7b64919ae6bca008c4efdee72).
Disabled and unsupported policies now remove `SIZE_RETENTION_BLOCKED` instead
of registering a zero-valued gauge. The gauge is absent for tables without size
retention and is also removed when an operator disables an existing policy.
Updated the enum documentation and tests, including a real-registry lifecycle
test that verifies the series is actually gone.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,173 @@ protected void processTable(String tableNameWithType) {
// Delete segments based on segment lineage and clean up segment lineage
metadata.
manageSegmentLineageCleanupForTable(tableConfig);
+
+ // Re-read active segments after time and lineage cleanup so
already-deleted bytes do not cause extra eviction.
+ manageSizeBasedRetention(tableConfig);
+ }
+
+ /// Enforces the optional compressed-segment size limit on live data
independently of time retention. Replaced
+ /// segments are excluded from accounting, and an in-progress replacement
postpones the entire pass. The newest
+ /// OFFLINE segment and realtime recovery segments are preserved; hybrid
realtime eviction requires offline coverage.
+ /// Unknown sizes or missing metadata disable this pass; consuming segments
are excluded from accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String tableNameWithType = tableConfig.getTableName();
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null || (tableConfig.getTableType() ==
TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig))))
{
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ return;
+ }
+ // Leave the gauge blocked on unsafe early returns or exceptions, and
clear it only after a successful pass.
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
Review Comment:
Fixed in
[c32ac7fa3a](https://github.com/apache/pinot/commit/c32ac7fa3aeb8cc7b64919ae6bca008c4efdee72).
The configured pass now computes its result locally and publishes the gauge
exactly once in `finally`. Healthy passes publish only 0, while skipped,
failed, or interrupted passes publish 1; the previous completed-pass value
remains visible during work. Regression assertions verify both the expected
terminal value and exactly one publication, including the exception path, and a
real metric registry verifies failure-to-retry recovery.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -122,6 +128,173 @@ protected void processTable(String tableNameWithType) {
// Delete segments based on segment lineage and clean up segment lineage
metadata.
manageSegmentLineageCleanupForTable(tableConfig);
+
+ // Re-read active segments after time and lineage cleanup so
already-deleted bytes do not cause extra eviction.
+ manageSizeBasedRetention(tableConfig);
+ }
+
+ /// Enforces the optional compressed-segment size limit on live data
independently of time retention. Replaced
+ /// segments are excluded from accounting, and an in-progress replacement
postpones the entire pass. The newest
+ /// OFFLINE segment and realtime recovery segments are preserved; hybrid
realtime eviction requires offline coverage.
+ /// Unknown sizes or missing metadata disable this pass; consuming segments
are excluded from accounting and eviction.
+ @VisibleForTesting
+ protected void manageSizeBasedRetention(TableConfig tableConfig) {
+ String tableNameWithType = tableConfig.getTableName();
+ String retentionSize =
tableConfig.getValidationConfig().getRetentionSize();
+ if (retentionSize == null || (tableConfig.getTableType() ==
TableType.OFFLINE
+ &&
!"APPEND".equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig))))
{
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ return;
+ }
+ // Leave the gauge blocked on unsafe early returns or exceptions, and
clear it only after a successful pass.
+ _controllerMetrics.setOrUpdateTableGauge(tableNameWithType,
ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
+ long retentionSizeBytes;
+ try {
+ retentionSizeBytes = DataSizeUtils.toBytes(retentionSize);
+ Preconditions.checkArgument(retentionSizeBytes > 0, "Retention size must
be positive");
+ } catch (Exception e) {
+ LOGGER.warn("Invalid retention size: {} for table: {}, skip size
retention", retentionSize, tableNameWithType);
+ return;
+ }
+
+ SegmentLineage segmentLineage =
+
SegmentLineageAccessHelper.getSegmentLineage(_pinotHelixResourceManager.getPropertyStore(),
tableNameWithType);
+ if (segmentLineage != null &&
segmentLineage.getLineageEntries().values().stream()
+ .anyMatch(entry -> entry.getState() == LineageEntryState.IN_PROGRESS))
{
+ LOGGER.warn("In-progress segment lineage for table: {}, skip size
retention", tableNameWithType);
Review Comment:
Addressed both points in
[c32ac7fa3a](https://github.com/apache/pinot/commit/c32ac7fa3aeb8cc7b64919ae6bca008c4efdee72).
The `retentionSize` Javadoc explicitly documents the legacy-mode race when
lineage-exclusive deletion is disabled. Selection now reuses one lineage
snapshot for the IN_PROGRESS check, replaced-segment accounting filter, and
candidate protection; the public deletion path keeps its fresh check when
enabled. IdealState is read before the shared lineage snapshot, preserving the
existing live-segment filtering order and preventing newly exposed replacement
destinations from being double-counted against an older lineage view.
Regression tests verify one planning lineage read and a replacement starting
during the IdealState read.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java:
##########
@@ -915,4 +923,577 @@ public void
testRetentionWithInvalidEndTimeAndCreationTimeFallback() {
// already asserts the correct segments via TestNG assertions
verify(pinotHelixResourceManager,
times(1)).deleteSegments(eq(OFFLINE_TABLE_NAME), anyList());
}
+
+ @Test
+ public void testSizeOnlyRetentionDeletesOldestSegmentsUntilWithinLimit() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("newest", 60, now),
+ createSizeRetentionSegment("oldest", 60, now - 2),
+ createSizeRetentionSegment("middle", 60, now - 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ // Exercise the complete table pass: a time-retention configuration is
deliberately absent.
+ retentionManager.processTable(OFFLINE_TABLE_NAME);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "middle"));
+ }
+
+ @Test
+ public void testSizeRetentionKeepsSegmentsAtOrBelowLimit() {
+ for (String limit : List.of("100B", "101B")) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, limit);
+ tableConfig.getValidationConfig().setReplication("3");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 1),
+ createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Stored archive sizes count once, regardless of table replication.
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionKeepsNewestOfflineSegmentWhenItAloneExceedsLimit() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("newest", 120, now),
+ createSizeRetentionSegment("oldest", 40, now - 1));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest"));
+ verify(metrics, atLeastOnce()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
+ verify(metrics, never()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ }
+
+ @Test
+ public void testSizeRetentionNeverDeletesOnlyOfflineSegment() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "1B");
+ List<SegmentZKMetadata> segments =
+ List.of(createSizeRetentionSegment("only", 100,
System.currentTimeMillis()));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verify(metrics, atLeastOnce()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
+ verify(metrics, never()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ }
+
+ @Test
+ public void testSizeRetentionOrderingWithCreationAndPushTimeFallbacks() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "20B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata creationFallback =
createSizeRetentionSegment("creationFallback", 10, -1);
+ creationFallback.setCreationTime(now - 1);
+ SegmentZKMetadata pushFallback =
createSizeRetentionSegment("pushFallback", 10, -1);
+ pushFallback.setCreationTime(-1);
+ pushFallback.setPushTime(now - 4);
+ List<SegmentZKMetadata> segments = List.of(creationFallback,
+ createSizeRetentionSegment("b", 10, now - 3), pushFallback,
+ createSizeRetentionSegment("oldest", 10, now - 5),
createSizeRetentionSegment("a", 10, now - 3));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(OFFLINE_TABLE_NAME,
List.of("oldest", "pushFallback", "a"));
+ }
+
+ @Test
+ public void
testSizeRetentionProtectsIncompleteAndLastCompletedRealtimeSegments() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.REALTIME, "50B");
+ long now = System.currentTimeMillis();
+ SegmentZKMetadata oldest = createSizeRetentionSegment(
+ new LLCSegmentName(TEST_TABLE_NAME, 0, 4, now - 3).getSegmentName(),
20, now - 3);
+ oldest.setStatus(Status.DONE);
+ SegmentZKMetadata lastCompleted = createSizeRetentionSegment(
+ new LLCSegmentName(TEST_TABLE_NAME, 0, 5, now - 1).getSegmentName(),
40, now - 1);
+ lastCompleted.setStatus(Status.DONE);
+ SegmentZKMetadata consuming = createSizeRetentionSegment("consuming", -1,
now - 5);
+ consuming.setStatus(Status.IN_PROGRESS);
+ SegmentZKMetadata committing = createSizeRetentionSegment("committing",
999, now - 4);
+ committing.setStatus(Status.COMMITTING);
+ SegmentZKMetadata uploaded = createSizeRetentionSegment("uploaded", 60,
now - 2);
+ List<SegmentZKMetadata> segments = List.of(lastCompleted, committing,
uploaded, consuming, oldest);
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).deleteSegments(REALTIME_TABLE_NAME,
List.of(oldest.getSegmentName(), "uploaded"));
+ }
+
+ @Test
+ public void testSizeRetentionSkipsTableWhenCompletedSegmentSizeIsUnknown() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 60, now - 2),
+ createSizeRetentionSegment("unknown", -1, now - 1),
createSizeRetentionSegment("newest", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verify(metrics, atLeastOnce()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
+ verify(metrics, never()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ }
+
+ @Test
+ public void testSizeRetentionIgnoresMetadataOutsideIdealState() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("oldest", 40, now - 2),
+ createSizeRetentionSegment("orphan", -1, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager);
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME,
true)).thenReturn(List.of("oldest", "newest"));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ }
+
+ @Test
+ public void
testSizeRetentionSkipsInProgressLineageEvenWhenExclusiveDeleteIsDisabled() {
+ for (boolean exclusiveDelete : List.of(true, false)) {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "50B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("olderLive", 60, now - 3),
+ createSizeRetentionSegment("source", 80, now - 2),
+ createSizeRetentionSegment("destination", 80, now - 1),
createSizeRetentionSegment("newestLive", 60, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(ControllerConf.LINEAGE_EXCLUSIVE_DELETE_ENABLED,
exclusiveDelete);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics,
+ conf, mock(BrokerServiceHelper.class));
+ SegmentLineage lineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ lineage.addLineageEntry("replacement",
+ new LineageEntry(List.of("source"), List.of("destination"),
LineageEntryState.IN_PROGRESS, now));
+ when(resourceManager.getPropertyStore().get(anyString(),
any(Stat.class), eq(AccessOption.PERSISTENT)))
+ .thenReturn(lineage.toZNRecord());
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ // Transient duplicate bytes must not trigger deletion of unrelated live
data.
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verify(resourceManager,
never()).deleteSegmentsForLineageCleanup(anyString(), anyList());
+ verify(resourceManager, never()).getSegmentsFor(anyString(),
anyBoolean());
+ verify(metrics, atLeastOnce()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
+ ControllerGauge.SIZE_RETENTION_BLOCKED, 1);
+ verify(metrics, never()).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
+ }
+ }
+
+ @Test
+ public void
testSizeRetentionExcludesCompletedReplacedSegmentsFromAccounting() {
+ TableConfig tableConfig =
createSizeRetentionTableConfig(TableType.OFFLINE, "100B");
+ long now = System.currentTimeMillis();
+ List<SegmentZKMetadata> segments =
List.of(createSizeRetentionSegment("source", -1, now - 2),
+ createSizeRetentionSegment("destination", 40, now - 1),
createSizeRetentionSegment("newest", 40, now));
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics metrics = mock(ControllerMetrics.class);
+ RetentionManager retentionManager =
createSizeRetentionManager(tableConfig, segments, resourceManager, metrics);
+ SegmentLineage lineage = new SegmentLineage(OFFLINE_TABLE_NAME);
+ lineage.addLineageEntry("replacement",
+ new LineageEntry(List.of("source"), List.of("destination"),
LineageEntryState.COMPLETED, now));
+ when(resourceManager.getPropertyStore().get(anyString(), any(Stat.class),
eq(AccessOption.PERSISTENT)))
+ .thenReturn(lineage.toZNRecord());
+ when(resourceManager.getSegmentsFor(OFFLINE_TABLE_NAME,
true)).thenReturn(List.of("destination", "newest"));
+
+ retentionManager.manageSizeBasedRetention(tableConfig);
+
+ verify(resourceManager).getSegmentsFor(OFFLINE_TABLE_NAME, true);
+ verify(resourceManager, never()).getSegmentsFor(OFFLINE_TABLE_NAME, false);
+ verify(resourceManager, never()).deleteSegments(anyString(), anyList());
+ verify(metrics).setOrUpdateTableGauge(OFFLINE_TABLE_NAME,
ControllerGauge.SIZE_RETENTION_BLOCKED, 0);
Review Comment:
Fixed in
[c32ac7fa3a](https://github.com/apache/pinot/commit/c32ac7fa3aeb8cc7b64919ae6bca008c4efdee72).
Success and blocked assertions now verify the expected gauge value plus
exactly one publication per pass, so an extra 1 after a successful 0 fails the
test. Added a real `ControllerMetrics`/`MetricValueUtils` lifecycle test that
checks actual registry values through healthy, rejected deletion, and retry
states, then verifies the gauge is absent after disabling retention and after
`nonLeaderCleanup`.
--
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]
