This is an automated email from the ASF dual-hosted git repository.
tarun11Mavani 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 5b53f30d801 Add controller support for rejecting out-of-retention
segment uploads (#18486)
5b53f30d801 is described below
commit 5b53f30d8016fed5da49488802ae280bb9425036
Author: Prabhat Singh <[email protected]>
AuthorDate: Thu Nov 12 18:54:33 2026 +0530
Add controller support for rejecting out-of-retention segment uploads
(#18486)
* Reject out-of-retention segment uploads (optional)
Add optional controller validation to reject offline segment uploads
outside the table retention window, align RetentionUtils segment end
time handling with minion raw time fields, emit a controller meter on
rejection, and extend tests and docs per review feedback.
* Drop trailing period on out-of-retention upload rejection message.
* Address review feedback on out-of-retention upload check
Have TableConfigRetentionUtils.buildRetentionStrategy reuse
RetentionUtils.parseTableDataRetentionMillis so the upload check and the
retention manager share one retention parser. Document that minion task
uploads are subject to the check and how a rejection interacts with the
RealtimeToOffline watermark, and clarify that the hybrid retention
strategy uses an independent boundary. Add TableConfigRetentionUtilsTest.
* Add integration test for out-of-retention segment upload rejection
Enable controller.segment.upload.rejectOutOfRetention.enabled in
SegmentUploadIntegrationTest and cover both outcomes end to end: an
upload outside a 7-day retention window returns 403, is not registered
or copied to deep store, and increments the rejection meter; an upload
inside the window succeeds without touching the meter.
* Address review feedback on out-of-retention upload check
- Rename controller.segment.upload.rejectOutOfRetention.enabled to
controller.segment.upload.reject.out.of.retention
- Drop the creation-time fallback option from the SegmentMetadata
isPurgeable
overload and restore the SegmentZKMetadata overload to its original body
- Make getSegmentMetadataEndTimeMillis private and cover it through
isPurgeable outcomes
- Rely on TableConfig's non-null validation config
- Use static Assert/Mockito imports in new test code; add class Javadoc
---
.../pinot/common/metrics/ControllerMeter.java | 4 +-
.../apache/pinot/common/utils/RetentionUtils.java | 73 ++++++++-
.../pinot/common/utils/RetentionUtilsTest.java | 164 +++++++++++++++++++++
.../apache/pinot/controller/ControllerConf.java | 23 +++
.../PinotSegmentUploadDownloadRestletResource.java | 2 +
.../api/upload/SegmentValidationUtils.java | 54 +++++++
.../helix/core/retention/RetentionManager.java | 8 +-
.../core/retention/TableConfigRetentionUtils.java | 19 +--
.../pinot/controller/ControllerConfTest.java | 17 +++
.../api/upload/SegmentValidationUtilsTest.java | 111 ++++++++++++++
.../retention/TableConfigRetentionUtilsTest.java | 94 ++++++++++++
.../tests/SegmentUploadIntegrationTest.java | 81 ++++++++++
12 files changed, 632 insertions(+), 18 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
index 02c87139355..83e40f4dfb7 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
@@ -102,7 +102,9 @@ public enum ControllerMeter implements
AbstractMetrics.Meter {
// Number of segment-delete requests rejected because the targets
participate in a live segment lineage entry.
LINEAGE_BLOCKED_DELETE_COUNT("LineageBlockedDeleteCount", false),
// Storage quota breach detected for a table whose enforcement mode is
ENFORCED or SHADOW (not DISABLED)
- STORAGE_QUOTA_EXCEEDED("StorageQuotaExceeded", false);
+ STORAGE_QUOTA_EXCEEDED("StorageQuotaExceeded", false),
+ // Segment uploads rejected for being outside the table retention window.
+
OUT_OF_RETENTION_SEGMENT_UPLOAD_REJECTED("OutOfRetentionSegmentUploadRejected",
true);
private final String _brokerMeterName;
private final String _unit;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/RetentionUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/RetentionUtils.java
index df5788b9596..aa2dbde1286 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/RetentionUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/RetentionUtils.java
@@ -18,19 +18,38 @@
*/
package org.apache.pinot.common.utils;
+import java.util.Locale;
+import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/// Utility methods for evaluating segment retention eligibility.
+/// Utility methods for evaluating segment retention eligibility and parsing
table data retention from config.
public class RetentionUtils {
private static final Logger LOGGER =
LoggerFactory.getLogger(RetentionUtils.class);
private RetentionUtils() {
}
+ /// Whether time-based data retention should run for this table's segments,
matching the controller retention manager:
+ /// for {@link TableType#OFFLINE} tables, retention applies only when batch
segment ingestion type is {@code APPEND}.
+ public static boolean shouldManageTimeBasedDataRetention(TableConfig
tableConfig) {
+ if (tableConfig.getTableType() == TableType.OFFLINE && !"APPEND"
+
.equalsIgnoreCase(IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig)))
{
+ return false;
+ }
+ return true;
+ }
+
/// Implements the time-comparison and creation-time fallback logic used by
`TimeRetentionStrategy` for
/// completed segments. Does NOT check segment completion status — callers
must guarantee that only completed
/// segments (DONE or UPLOADED) are passed in.
@@ -69,4 +88,56 @@ public class RetentionUtils {
}
return false;
}
+
+ /// Whether segment file metadata is past the retention window, using the
end time from
+ /// {@link SegmentMetadata#getEndTime()} converted with {@link
SegmentMetadata#getTimeUnit()}. Unlike the
+ /// {@link SegmentZKMetadata} overload, there is no creation-time fallback.
+ ///
+ /// - If end time is valid: purgeable when `currentTimeMs - endTimeMs >
retentionMs`.
+ /// - Otherwise: not purgeable (fail-open).
+ public static boolean isPurgeable(String tableNameWithType, SegmentMetadata
segmentMetadata, long retentionMs,
+ long currentTimeMs) {
+ long endTimeMs = getSegmentMetadataEndTimeMillis(segmentMetadata);
+ if (TimeUtils.timeValueInValidRange(endTimeMs)) {
+ return currentTimeMs - endTimeMs > retentionMs;
+ }
+ LOGGER.warn("Segment: {} of table: {} has invalid end time in millis: {}.
Cannot determine retention, skipping",
+ segmentMetadata.getName(), tableNameWithType, endTimeMs);
+ return false;
+ }
+
+ /// Parses {@link
SegmentsValidationAndRetentionConfig#getRetentionTimeUnit()} and
+ /// {@link SegmentsValidationAndRetentionConfig#getRetentionTimeValue()}
into table data retention duration in
+ /// milliseconds (same interpretation as controller time-based retention).
+ ///
+ /// @return millis, or empty if unset or not parseable
+ public static OptionalLong
parseTableDataRetentionMillis(SegmentsValidationAndRetentionConfig
validationConfig) {
+ String retentionTimeUnit = validationConfig.getRetentionTimeUnit();
+ String retentionTimeValue = validationConfig.getRetentionTimeValue();
+ if (StringUtils.isEmpty(retentionTimeUnit) ||
StringUtils.isEmpty(retentionTimeValue)) {
+ return OptionalLong.empty();
+ }
+ try {
+ return
OptionalLong.of(TimeUnit.valueOf(retentionTimeUnit.toUpperCase(Locale.ROOT))
+ .toMillis(Long.parseLong(retentionTimeValue)));
+ } catch (Exception e) {
+ LOGGER.debug("Failed to parse table data retention: unit='{}'
value='{}'", retentionTimeUnit, retentionTimeValue,
+ e);
+ return OptionalLong.empty();
+ }
+ }
+
+ /// Returns segment end time in epoch millis from on-disk {@link
SegmentMetadata} for retention comparison, using
+ /// {@link SegmentMetadata#getTimeUnit()} and {@link
SegmentMetadata#getEndTime()} (same raw fields minion task
+ /// executors pass when configuring generation from segment files). When
`timeUnit` is non-null and raw end time is
+ /// not {@link Long#MIN_VALUE}, returns `timeUnit.toMillis(endTime)`;
otherwise returns the raw end time (typically
+ /// `Long.MIN_VALUE` when unset).
+ private static long getSegmentMetadataEndTimeMillis(SegmentMetadata
segmentMetadata) {
+ TimeUnit timeUnit = segmentMetadata.getTimeUnit();
+ long endTime = segmentMetadata.getEndTime();
+ if (timeUnit != null && endTime != Long.MIN_VALUE) {
+ return timeUnit.toMillis(endTime);
+ }
+ return endTime;
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/RetentionUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/RetentionUtilsTest.java
index 08ce605f5ff..b0822336067 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/RetentionUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/RetentionUtilsTest.java
@@ -18,10 +18,20 @@
*/
package org.apache.pinot.common.utils;
+import java.util.Locale;
+import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
@@ -47,6 +57,14 @@ public class RetentionUtilsTest {
return segment;
}
+ private static SegmentMetadata mockSegmentMetadata(TimeUnit timeUnit, long
endTime) {
+ SegmentMetadata metadata = mock(SegmentMetadata.class);
+ when(metadata.getName()).thenReturn("seg");
+ when(metadata.getTimeUnit()).thenReturn(timeUnit);
+ when(metadata.getEndTime()).thenReturn(endTime);
+ return metadata;
+ }
+
@Test
public void testExpiredSegmentIsPurgeable() {
long now = System.currentTimeMillis();
@@ -127,4 +145,150 @@ public class RetentionUtilsTest {
SegmentZKMetadata segment = makeSegmentWithCreationTime(now - 2 *
ONE_DAY_MS, now - 10 * ONE_DAY_MS);
assertFalse(RetentionUtils.isPurgeable(TABLE_NAME, segment, RETENTION_MS,
now, true));
}
+
+ @Test
+ public void testSegmentMetadataExpiredSegmentIsPurgeable() {
+ long now = System.currentTimeMillis();
+ assertTrue(RetentionUtils.isPurgeable(TABLE_NAME,
+ mockSegmentMetadata(TimeUnit.MILLISECONDS, now - 10 * ONE_DAY_MS),
RETENTION_MS, now));
+ }
+
+ @Test
+ public void testSegmentMetadataExactBoundaryIsNotPurgeable() {
+ long now = System.currentTimeMillis();
+ assertFalse(RetentionUtils.isPurgeable(TABLE_NAME,
mockSegmentMetadata(TimeUnit.MILLISECONDS, now - RETENTION_MS),
+ RETENTION_MS, now));
+ }
+
+ @Test
+ public void testSegmentMetadataOneMsPastBoundaryIsPurgeable() {
+ long now = System.currentTimeMillis();
+ assertTrue(RetentionUtils.isPurgeable(TABLE_NAME,
+ mockSegmentMetadata(TimeUnit.MILLISECONDS, now - RETENTION_MS - 1),
RETENTION_MS, now));
+ }
+
+ @Test
+ public void testSegmentMetadataEndTimeConvertedWithTimeUnit() {
+ long nowDays = 20_000;
+ long now = TimeUnit.DAYS.toMillis(nowDays);
+ assertTrue(RetentionUtils.isPurgeable(TABLE_NAME,
mockSegmentMetadata(TimeUnit.DAYS, nowDays - 8), RETENTION_MS,
+ now));
+ assertFalse(RetentionUtils.isPurgeable(TABLE_NAME,
mockSegmentMetadata(TimeUnit.DAYS, nowDays - 6), RETENTION_MS,
+ now));
+ }
+
+ @Test
+ public void testSegmentMetadataInvalidEndTimeIsNotPurgeable() {
+ long now = System.currentTimeMillis();
+ assertFalse(RetentionUtils.isPurgeable(TABLE_NAME,
mockSegmentMetadata(TimeUnit.MILLISECONDS, -1), RETENTION_MS,
+ now));
+ }
+
+ @Test
+ public void testSegmentMetadataWithoutTimeInfoIsNotPurgeable() {
+ // Segments without time metadata have no time unit and an unset end time
+ long now = System.currentTimeMillis();
+ assertFalse(RetentionUtils.isPurgeable(TABLE_NAME,
mockSegmentMetadata(null, Long.MIN_VALUE), RETENTION_MS, now));
+ }
+
+ @Test
+ public void shouldManageTimeBasedDataRetentionRealtime() {
+ TableConfig cfg = new
TableConfigBuilder(TableType.REALTIME).setTableName("t").build();
+ assertTrue(RetentionUtils.shouldManageTimeBasedDataRetention(cfg));
+ }
+
+ @Test
+ public void shouldManageTimeBasedDataRetentionOfflineAppendDefault() {
+ TableConfig cfg = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t_OFFLINE").build();
+ assertTrue(RetentionUtils.shouldManageTimeBasedDataRetention(cfg));
+ }
+
+ @Test
+ public void shouldManageTimeBasedDataRetentionOfflineRefreshSkipped() {
+ TableConfig cfg = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t_OFFLINE").setSegmentPushType("REFRESH")
+ .build();
+ assertFalse(RetentionUtils.shouldManageTimeBasedDataRetention(cfg));
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisValidDays() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("DAYS");
+ cfg.setRetentionTimeValue("7");
+ assertTrue(RetentionUtils.parseTableDataRetentionMillis(cfg).isPresent());
+
assertEquals(RetentionUtils.parseTableDataRetentionMillis(cfg).getAsLong(),
TimeUnit.DAYS.toMillis(7));
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisMissingReturnsEmpty() {
+ assertFalse(RetentionUtils.parseTableDataRetentionMillis(new
SegmentsValidationAndRetentionConfig()).isPresent());
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisInvalidUnitReturnsEmpty() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("FORTNIGHTS");
+ cfg.setRetentionTimeValue("1");
+ assertFalse(RetentionUtils.parseTableDataRetentionMillis(cfg).isPresent());
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisEmptyUnitReturnsEmpty() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("");
+ cfg.setRetentionTimeValue("7");
+ assertFalse(RetentionUtils.parseTableDataRetentionMillis(cfg).isPresent());
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisEmptyValueReturnsEmpty() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("DAYS");
+ cfg.setRetentionTimeValue("");
+ assertFalse(RetentionUtils.parseTableDataRetentionMillis(cfg).isPresent());
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisInvalidValueReturnsEmpty() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("DAYS");
+ cfg.setRetentionTimeValue("not-a-number");
+ assertFalse(RetentionUtils.parseTableDataRetentionMillis(cfg).isPresent());
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisLowerCaseUnitParses() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("days");
+ cfg.setRetentionTimeValue("7");
+ OptionalLong parsed = RetentionUtils.parseTableDataRetentionMillis(cfg);
+ assertTrue(parsed.isPresent());
+ assertEquals(parsed.getAsLong(), TimeUnit.DAYS.toMillis(7));
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisLowerCaseUnitUsesLocaleRoot() {
+ Locale previous = Locale.getDefault();
+ try {
+ Locale.setDefault(Locale.forLanguageTag("tr-TR"));
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("minutes");
+ cfg.setRetentionTimeValue("5");
+ OptionalLong parsed = RetentionUtils.parseTableDataRetentionMillis(cfg);
+ assertTrue(parsed.isPresent());
+ assertEquals(parsed.getAsLong(), TimeUnit.MINUTES.toMillis(5));
+ } finally {
+ Locale.setDefault(previous);
+ }
+ }
+
+ @Test
+ public void testParseTableDataRetentionMillisValidHours() {
+ SegmentsValidationAndRetentionConfig cfg = new
SegmentsValidationAndRetentionConfig();
+ cfg.setRetentionTimeUnit("HOURS");
+ cfg.setRetentionTimeValue("12");
+ OptionalLong parsed = RetentionUtils.parseTableDataRetentionMillis(cfg);
+ assertTrue(parsed.isPresent());
+ assertEquals(parsed.getAsLong(), TimeUnit.HOURS.toMillis(12));
+ }
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
index fd6ccdcdea5..cd63977034d 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
@@ -370,6 +370,17 @@ public class ControllerConf extends PinotConfiguration {
// Amount of the time the segment can take from the beginning of upload to
the end of upload. Used when parallel push
// protection is enabled. If the upload does not finish within the timeout,
next upload can override the previous one.
public static final String SEGMENT_UPLOAD_TIMEOUT_IN_MILLIS =
"controller.segment.upload.timeoutInMillis";
+ /// If true, the controller may reject single-segment uploads with HTTP 403
when segment data end is past the table
+ /// data retention window (see {@link
org.apache.pinot.controller.api.upload.SegmentValidationUtils}).
+ ///
+ /// This also applies to minion task uploads that push through `POST
/segments`: always for Purge, UpsertCompaction
+ /// and RefreshSegment, and for RealtimeToOffline, MergeRollup and
UpsertCompactMerge unless `batchSegmentUpload` is
+ /// on.
+ /// A rejected RealtimeToOffline upload does not advance its watermark, so
the task retries the same window until the
+ /// source segments age out (indefinitely with the hybrid retention
strategy, since the time boundary cannot advance).
+ /// Review task schedules and buffer periods against table retention before
enabling.
+ public static final String SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION =
+ "controller.segment.upload.reject.out.of.retention";
public static final String REALTIME_SEGMENT_METADATA_COMMIT_NUMLOCKS =
"controller.realtime.segment.metadata.commit.numLocks";
public static final String ENABLE_STORAGE_QUOTA_CHECK =
"controller.enable.storage.quota.check";
@@ -419,6 +430,8 @@ public class ControllerConf extends PinotConfiguration {
public static final String DEFAULT_REBALANCE_PRE_CHECKER =
"org.apache.pinot.controller.helix.core.rebalance.DefaultRebalancePreChecker";
public static final long DEFAULT_SEGMENT_UPLOAD_TIMEOUT_IN_MILLIS =
600_000L; // 10 minutes
+ /// Default for {@link #SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION}: disabled
until explicitly enabled.
+ public static final boolean DEFAULT_SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION =
false;
public static final int DEFAULT_MIN_NUM_CHARS_IN_IS_TO_TURN_ON_COMPRESSION =
-1;
public static final int DEFAULT_REALTIME_SEGMENT_METADATA_COMMIT_NUMLOCKS =
64;
public static final boolean DEFAULT_ENABLE_STORAGE_QUOTA_CHECK = true;
@@ -1254,6 +1267,16 @@ public class ControllerConf extends PinotConfiguration {
setProperty(SEGMENT_UPLOAD_TIMEOUT_IN_MILLIS,
segmentUploadTimeoutInMillis);
}
+ /// @return whether out-of-retention single-segment upload rejection is
enabled
+ public boolean isSegmentUploadRejectOutOfRetention() {
+ return getProperty(SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION,
DEFAULT_SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION);
+ }
+
+ /// @param rejectOutOfRetention value for {@link
#SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION}
+ public void setSegmentUploadRejectOutOfRetention(boolean
rejectOutOfRetention) {
+ setProperty(SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION, rejectOutOfRetention);
+ }
+
public int getRealtimeSegmentMetadataCommitNumLocks() {
return getProperty(REALTIME_SEGMENT_METADATA_COMMIT_NUMLOCKS,
DEFAULT_REALTIME_SEGMENT_METADATA_COMMIT_NUMLOCKS);
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
index d90164e46c7..f269cc212cb 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java
@@ -389,6 +389,8 @@ public class PinotSegmentUploadDownloadRestletResource {
SegmentValidationUtils.validateTimeInterval(segmentMetadata,
tableConfig);
}
SegmentValidationUtils.validateUpsertSegmentPartitionMetadata(segmentMetadata,
tableConfig);
+ SegmentValidationUtils.rejectUploadIfOutOfRetention(segmentMetadata,
tableConfig, System.currentTimeMillis(),
+ _controllerConf.isSegmentUploadRejectOutOfRetention(),
_controllerMetrics);
long untarredSegmentSizeInBytes;
if (uploadType == FileUploadType.METADATA && segmentSizeInBytes > 0) {
// TODO: Include the untarred segment size when using the METADATA
push rest API. Currently we can only use the
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
index 7c5875d59ff..f0a0c804975 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidationUtils.java
@@ -19,9 +19,13 @@
package org.apache.pinot.controller.api.upload;
import java.util.Map;
+import java.util.OptionalLong;
import java.util.Set;
import javax.annotation.Nullable;
import javax.ws.rs.core.Response;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.utils.RetentionUtils;
import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.validation.StorageQuotaChecker;
import org.apache.pinot.segment.local.utils.TableConfigUtils;
@@ -29,6 +33,7 @@ import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.spi.config.table.QuotaEnforcementMode;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
+import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.TimeUtils;
@@ -103,6 +108,55 @@ public class SegmentValidationUtils {
return null;
}
+ /// When `controller.segment.upload.reject.out.of.retention` is true on the
controller, rejects the upload if
+ /// the segment's data end time (from segment file metadata) is past the
table's `retentionTimeValue` /
+ /// `retentionTimeUnit` window. This matches the retention manager's
time-based boundary. It does not model the
+ /// hybrid retention strategy
(`controller.enable.hybrid.table.retention.strategy`), under which the REALTIME
side of
+ /// a hybrid table is purged by the offline time boundary instead of the
retention window. The two boundaries are
+ /// independent there, so this check can be either stricter or looser than
what the retention manager purges.
+ /// **Index creation time is not** used as a fallback here: it can reflect
upstream segment timestamps (for example
+ /// upsert compaction) and is a poor proxy for the data window at upload
time.
+ ///
+ /// Rejection uses HTTP 403 so callers that enable the controller flag can
treat the response as a hard failure.
+ /// Clients that do not enable the flag see no behavior change.
+ ///
+ /// Controller wiring: {@link
org.apache.pinot.controller.api.resources.PinotSegmentUploadDownloadRestletResource}
+ /// invokes this for single-segment upload only. METADATA batch upload
(`POST /segments/batchUpload`) and
+ /// reingested-segment upload do not call it.
+ ///
+ /// Minion task uploads are not exempt. Purge, UpsertCompaction and
RefreshSegment always push through this
+ /// single-segment endpoint. RealtimeToOffline, MergeRollup and
UpsertCompactMerge do too unless `batchSegmentUpload`
+ /// is enabled for the task. A rejected RealtimeToOffline upload does not
advance the task watermark, so the same
+ /// window is retried on every run until the source segments age out.
+ ///
+ /// For tables where the retention manager does not apply time-based
retention to completed segments (offline tables
+ /// whose batch ingestion type is not `APPEND`), this method returns without
evaluating retention.
+ public static void rejectUploadIfOutOfRetention(SegmentMetadata
segmentMetadata, TableConfig tableConfig,
+ long currentTimeMs, boolean rejectOutOfRetention, @Nullable
ControllerMetrics controllerMetrics) {
+ if (!rejectOutOfRetention) {
+ return;
+ }
+ if (!RetentionUtils.shouldManageTimeBasedDataRetention(tableConfig)) {
+ return;
+ }
+ SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
+ OptionalLong retentionMsOpt =
RetentionUtils.parseTableDataRetentionMillis(validationConfig);
+ if (retentionMsOpt.isEmpty()) {
+ return;
+ }
+ long retentionMs = retentionMsOpt.getAsLong();
+ if (RetentionUtils.isPurgeable(tableConfig.getTableName(),
segmentMetadata, retentionMs, currentTimeMs)) {
+ if (controllerMetrics != null) {
+
controllerMetrics.addMeteredGlobalValue(ControllerMeter.OUT_OF_RETENTION_SEGMENT_UPLOAD_REJECTED,
1L);
+ }
+ throw new ControllerApplicationException(LOGGER, String.format(
+ "Segment %s of table %s is outside the retention window (%s %s);
upload rejected",
+ segmentMetadata.getName(), tableConfig.getTableName(),
validationConfig.getRetentionTimeValue(),
+ validationConfig.getRetentionTimeUnit()),
+ Response.Status.FORBIDDEN);
+ }
+ }
+
public static void checkStorageQuota(String segmentName, long
tarSegmentSizeInBytes, long untarredSegmentSizeInBytes,
TableConfig tableConfig,
StorageQuotaChecker quotaChecker) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
index 5b3f4d4d291..92547929d14 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
@@ -45,6 +45,7 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.utils.RetentionUtils;
import org.apache.pinot.common.utils.TarCompressionUtils;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.ControllerConf;
@@ -333,11 +334,10 @@ public class RetentionManager extends
ControllerPeriodicTask<Void> {
String tableNameWithType = tableConfig.getTableName();
LOGGER.info("Start managing retention for table: {}", tableNameWithType);
- // For offline tables, ensure that the segmentPushType is APPEND.
SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
- String segmentPushType =
IngestionConfigUtils.getBatchSegmentIngestionType(tableConfig);
- if (tableConfig.getTableType() == TableType.OFFLINE &&
!"APPEND".equalsIgnoreCase(segmentPushType)) {
- LOGGER.info("Segment push type is not APPEND for table: {}, skip
managing retention", tableNameWithType);
+ if (!RetentionUtils.shouldManageTimeBasedDataRetention(tableConfig)) {
+ LOGGER.info("Skipping time-based retention for table: {} (offline tables
require APPEND batch ingestion)",
+ tableNameWithType);
return;
}
int untrackedSegmentsDeletionBatchSize =
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtils.java
index 9f1bf991a9d..439be7fefca 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtils.java
@@ -18,9 +18,10 @@
*/
package org.apache.pinot.controller.helix.core.retention;
-import java.util.Locale;
+import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.RetentionUtils;
import
org.apache.pinot.controller.helix.core.retention.strategy.RetentionStrategy;
import
org.apache.pinot.controller.helix.core.retention.strategy.TimeRetentionStrategy;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
@@ -46,18 +47,12 @@ public class TableConfigRetentionUtils {
public static RetentionStrategy buildRetentionStrategy(TableConfig
tableConfig,
boolean useCreationTimeFallback) {
SegmentsValidationAndRetentionConfig validationConfig =
tableConfig.getValidationConfig();
- String unit = validationConfig.getRetentionTimeUnit();
- String value = validationConfig.getRetentionTimeValue();
- if (unit == null || unit.isEmpty() || value == null || value.isEmpty()) {
- LOGGER.warn("Invalid retention time: {} {} for table: {}, skip", unit,
value, tableConfig.getTableName());
- return null;
- }
- try {
- return new
TimeRetentionStrategy(TimeUnit.valueOf(unit.toUpperCase(Locale.ROOT)),
Long.parseLong(value),
- useCreationTimeFallback);
- } catch (Exception e) {
- LOGGER.warn("Invalid retention time: {} {} for table: {}, skip", unit,
value, tableConfig.getTableName());
+ OptionalLong retentionMs =
RetentionUtils.parseTableDataRetentionMillis(validationConfig);
+ if (retentionMs.isEmpty()) {
+ LOGGER.warn("Invalid retention time: {} {} for table: {}, skip",
validationConfig.getRetentionTimeUnit(),
+ validationConfig.getRetentionTimeValue(),
tableConfig.getTableName());
return null;
}
+ return new TimeRetentionStrategy(TimeUnit.MILLISECONDS,
retentionMs.getAsLong(), useCreationTimeFallback);
}
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerConfTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerConfTest.java
index 1fc53dcd1eb..8c4b07fbefc 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerConfTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerConfTest.java
@@ -32,6 +32,8 @@ import org.testng.Assert;
import org.testng.annotations.Test;
import static
org.apache.pinot.controller.ControllerConf.ControllerPeriodicTasksConf.*;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
public class ControllerConfTest {
@@ -296,4 +298,19 @@ public class ControllerConfTest {
Assert.assertEquals(conf.getSegmentReplaceExternalViewCheckIntervalMs(),
500L);
Assert.assertEquals(conf.getSegmentReplaceMaxRetryAttempts(), 2);
}
+
+ @Test
+ public void testSegmentUploadRejectOutOfRetentionDefault() {
+ ControllerConf conf = new ControllerConf();
+ assertFalse(conf.isSegmentUploadRejectOutOfRetention());
+ }
+
+ @Test
+ public void testSegmentUploadRejectOutOfRetentionOverride() {
+ ControllerConf conf = new ControllerConf();
+ conf.setSegmentUploadRejectOutOfRetention(true);
+ assertTrue(conf.isSegmentUploadRejectOutOfRetention());
+ conf.setSegmentUploadRejectOutOfRetention(false);
+ assertFalse(conf.isSegmentUploadRejectOutOfRetention());
+ }
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
index 6ef9f03d28a..93cc38158de 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/SegmentValidationUtilsTest.java
@@ -20,7 +20,10 @@ package org.apache.pinot.controller.api.upload;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import javax.ws.rs.core.Response;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.validation.StorageQuotaChecker;
import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -40,6 +43,7 @@ import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertThrows;
@@ -140,6 +144,105 @@ public class SegmentValidationUtilsTest {
SegmentValidationUtils.validateUpsertSegmentPartitionMetadata(mockSegmentMetadata(Set.of(1)),
tableConfig);
}
+ @Test
+ public void testRejectUploadSkippedWhenDisabled() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.OFFLINE);
+
SegmentValidationUtils.rejectUploadIfOutOfRetention(mockRetentionSegmentMetadata(now
- TimeUnit.DAYS.toMillis(30)),
+ tableConfig, now, false, null);
+ }
+
+ @Test
+ public void testRejectUploadSkippedForOfflineRefresh() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(RAW_TABLE_NAME)
+ .setRetentionTimeUnit("DAYS")
+ .setRetentionTimeValue("7")
+ .setSegmentPushType("REFRESH")
+ .build();
+
SegmentValidationUtils.rejectUploadIfOutOfRetention(mockRetentionSegmentMetadata(now
- TimeUnit.DAYS.toMillis(30)),
+ tableConfig, now, true, null);
+ }
+
+ @Test
+ public void testRejectUploadSkippedWhenRetentionInvalid() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(RAW_TABLE_NAME)
+ .setRetentionTimeUnit("BAD")
+ .setRetentionTimeValue("7")
+ .build();
+
SegmentValidationUtils.rejectUploadIfOutOfRetention(mockRetentionSegmentMetadata(now
- TimeUnit.DAYS.toMillis(30)),
+ tableConfig, now, true, null);
+ }
+
+ @Test
+ public void testRejectUploadWhenEnabledAndSegmentOutOfRetention() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.OFFLINE);
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
+ () -> SegmentValidationUtils.rejectUploadIfOutOfRetention(
+ mockRetentionSegmentMetadata(now - TimeUnit.DAYS.toMillis(30)),
tableConfig, now, true,
+ controllerMetrics));
+ assertEquals(exception.getResponse().getStatus(),
Response.Status.FORBIDDEN.getStatusCode());
+
verify(controllerMetrics).addMeteredGlobalValue(ControllerMeter.OUT_OF_RETENTION_SEGMENT_UPLOAD_REJECTED,
1L);
+ }
+
+ @Test
+ public void testRejectUploadAllowsRecentSegment() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.OFFLINE);
+
SegmentValidationUtils.rejectUploadIfOutOfRetention(mockRetentionSegmentMetadata(now
- TimeUnit.DAYS.toMillis(1)),
+ tableConfig, now, true, null);
+ }
+
+ @Test
+ public void testRejectUploadDoesNotUseIndexCreationTimeFallback() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.OFFLINE);
+ SegmentMetadata segmentMetadata = mockRetentionSegmentMetadata(-1);
+ when(segmentMetadata.getIndexCreationTime()).thenReturn(now -
TimeUnit.DAYS.toMillis(30));
+ SegmentValidationUtils.rejectUploadIfOutOfRetention(segmentMetadata,
tableConfig, now, true, null);
+ }
+
+ @Test
+ public void testRejectUploadSkippedWhenRetentionNotConfigured() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+
SegmentValidationUtils.rejectUploadIfOutOfRetention(mockRetentionSegmentMetadata(now
- TimeUnit.DAYS.toMillis(30)),
+ tableConfig, now, true, null);
+ }
+
+ @Test
+ public void testRejectUploadThrowsWhenOutOfRetentionAndMetricsNull() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.OFFLINE);
+ ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
+ () -> SegmentValidationUtils.rejectUploadIfOutOfRetention(
+ mockRetentionSegmentMetadata(now - TimeUnit.DAYS.toMillis(30)),
tableConfig, now, true, null));
+ assertEquals(exception.getResponse().getStatus(),
Response.Status.FORBIDDEN.getStatusCode());
+ }
+
+ @Test
+ public void testRejectUploadRealtimeOutOfRetentionThrows() {
+ long now = System.currentTimeMillis();
+ TableConfig tableConfig = getRetentionTableConfig(TableType.REALTIME);
+ ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
+ () -> SegmentValidationUtils.rejectUploadIfOutOfRetention(
+ mockRetentionSegmentMetadata(now - TimeUnit.DAYS.toMillis(30)),
tableConfig, now, true, null));
+ assertEquals(exception.getResponse().getStatus(),
Response.Status.FORBIDDEN.getStatusCode());
+ }
+
+ private static TableConfig getRetentionTableConfig(TableType tableType) {
+ return new TableConfigBuilder(tableType)
+ .setTableName(RAW_TABLE_NAME)
+ .setRetentionTimeUnit("DAYS")
+ .setRetentionTimeValue("7")
+ .build();
+ }
+
private static TableConfig getOfflineUpsertTableConfig() {
return new TableConfigBuilder(TableType.OFFLINE)
.setTableName(RAW_TABLE_NAME)
@@ -173,6 +276,14 @@ public class SegmentValidationUtilsTest {
return segmentMetadata;
}
+ private static SegmentMetadata mockRetentionSegmentMetadata(long endTimeMs) {
+ SegmentMetadata segmentMetadata = mock(SegmentMetadata.class);
+ when(segmentMetadata.getTimeUnit()).thenReturn(TimeUnit.MILLISECONDS);
+ when(segmentMetadata.getEndTime()).thenReturn(endTimeMs);
+ when(segmentMetadata.getName()).thenReturn(SEGMENT_NAME);
+ return segmentMetadata;
+ }
+
private static ColumnMetadata mockColumnMetadata(Set<Integer> partitions) {
ColumnMetadata columnMetadata = mock(ColumnMetadata.class);
when(columnMetadata.getPartitions()).thenReturn(partitions);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtilsTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtilsTest.java
new file mode 100644
index 00000000000..61c1484d593
--- /dev/null
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/TableConfigRetentionUtilsTest.java
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.retention;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import
org.apache.pinot.controller.helix.core.retention.strategy.RetentionStrategy;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests for {@link TableConfigRetentionUtils#buildRetentionStrategy}: the
strategy purges segments older than the
+/// table retention window, accepts case-insensitive time units, honors the
creation-time fallback flag, and is not
+/// built when the retention unit or value is missing or invalid.
+public class TableConfigRetentionUtilsTest {
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+
+ @Test
+ public void testBuildRetentionStrategyUsesConfiguredWindow() {
+ RetentionStrategy strategy =
TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("DAYS", "7"),
false);
+ assertNotNull(strategy);
+ long now = System.currentTimeMillis();
+ String tableName = RAW_TABLE_NAME + "_OFFLINE";
+ assertTrue(strategy.isPurgeable(tableName, SEGMENT_NAME, now -
TimeUnit.DAYS.toMillis(8)));
+ assertFalse(strategy.isPurgeable(tableName, SEGMENT_NAME, now -
TimeUnit.DAYS.toMillis(6)));
+ }
+
+ @Test
+ public void testBuildRetentionStrategyAcceptsLowercaseUnit() {
+ RetentionStrategy strategy =
TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("hours", "2"),
false);
+ assertNotNull(strategy);
+ long now = System.currentTimeMillis();
+ String tableName = RAW_TABLE_NAME + "_OFFLINE";
+ assertTrue(strategy.isPurgeable(tableName, SEGMENT_NAME, now -
TimeUnit.HOURS.toMillis(3)));
+ assertFalse(strategy.isPurgeable(tableName, SEGMENT_NAME, now -
TimeUnit.HOURS.toMillis(1)));
+ }
+
+ @Test
+ public void testBuildRetentionStrategyPropagatesCreationTimeFallback() {
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ segmentZKMetadata.setCreationTime(System.currentTimeMillis() -
TimeUnit.DAYS.toMillis(30));
+ String tableName = RAW_TABLE_NAME + "_OFFLINE";
+
+ RetentionStrategy withFallback =
+
TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("DAYS", "7"),
true);
+ assertNotNull(withFallback);
+ assertTrue(withFallback.isPurgeable(tableName, segmentZKMetadata));
+
+ RetentionStrategy withoutFallback =
+
TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("DAYS", "7"),
false);
+ assertNotNull(withoutFallback);
+ assertFalse(withoutFallback.isPurgeable(tableName, segmentZKMetadata));
+ }
+
+ @Test
+ public void
testBuildRetentionStrategyReturnsNullForInvalidOrMissingRetention() {
+
assertNull(TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("BAD",
"7"), false));
+
assertNull(TableConfigRetentionUtils.buildRetentionStrategy(getTableConfig("DAYS",
"abc"), false));
+ assertNull(TableConfigRetentionUtils.buildRetentionStrategy(
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build(),
false));
+ }
+
+ private static TableConfig getTableConfig(String retentionTimeUnit, String
retentionTimeValue) {
+ return new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(RAW_TABLE_NAME)
+ .setRetentionTimeUnit(retentionTimeUnit)
+ .setRetentionTimeValue(retentionTimeValue)
+ .build();
+ }
+}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
index 1194dd1c6aa..81f3549e96e 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
@@ -22,10 +22,16 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
+import java.net.URI;
import java.util.List;
import java.util.Map;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hc.core5.http.HttpStatus;
+import org.apache.pinot.common.exception.HttpErrorStatusException;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.utils.FileUploadDownloadClient;
+import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.plugin.ingestion.batch.common.BaseSegmentPushJobRunner;
import
org.apache.pinot.plugin.ingestion.batch.standalone.SegmentMetadataPushJobRunner;
import
org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner;
@@ -39,6 +45,7 @@ import org.apache.pinot.spi.ingestion.batch.spec.PinotFSSpec;
import org.apache.pinot.spi.ingestion.batch.spec.PushJobSpec;
import org.apache.pinot.spi.ingestion.batch.spec.SegmentGenerationJobSpec;
import org.apache.pinot.spi.ingestion.batch.spec.TableSpec;
+import org.apache.pinot.spi.metrics.PinotMeter;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.util.TestUtils;
@@ -49,6 +56,12 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
/// Test for advanced push types.
/// Currently only tests METADATA push type.
@@ -93,6 +106,13 @@ public class SegmentUploadIntegrationTest extends
BaseClusterIntegrationTest {
TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
}
+ @Override
+ protected void overrideControllerConf(Map<String, Object> properties) {
+ super.overrideControllerConf(properties);
+ // Applies to every test in this class: any APPEND table configured with a
retention window is checked on upload.
+ properties.put(ControllerConf.SEGMENT_UPLOAD_REJECT_OUT_OF_RETENTION,
true);
+ }
+
@BeforeClass
public void setUp()
throws Exception {
@@ -390,6 +410,67 @@ public class SegmentUploadIntegrationTest extends
BaseClusterIntegrationTest {
Assert.assertTrue(segmentLineageResponse.contains("\"segmentsTo\":[\"" +
secondSegmentName + "\"]"));
}
+ @Test
+ public void testUploadRejectedWhenSegmentOutOfRetention()
+ throws Exception {
+ // The test data ends in 2014, so it is far outside a 7-day retention
window.
+ File segmentTarFile = createTableAndBuildSegment("7");
+ long rejectedBefore = getOutOfRetentionRejectedCount();
+
+ try (FileUploadDownloadClient fileUploadDownloadClient = new
FileUploadDownloadClient()) {
+ HttpErrorStatusException exception =
expectThrows(HttpErrorStatusException.class,
+ () ->
fileUploadDownloadClient.uploadSegment(URI.create(getOrCreateAdminClient().getSegmentUploadUrl()),
+ segmentTarFile.getName(), segmentTarFile, getTableName(),
TableType.OFFLINE));
+ assertEquals(exception.getStatusCode(), HttpStatus.SC_FORBIDDEN);
+ assertTrue(exception.getMessage().contains("outside the retention
window"), exception.getMessage());
+ }
+
+ assertEquals(getSegmentsList().size(), 0);
+ assertFalse(new File(_controllerConfig.getDataDir(),
getTableName()).exists());
+ assertEquals(getOutOfRetentionRejectedCount(), rejectedBefore + 1);
+ }
+
+ @Test
+ public void testUploadAcceptedWhenSegmentWithinRetention()
+ throws Exception {
+ File segmentTarFile = createTableAndBuildSegment("36500");
+ long rejectedBefore = getOutOfRetentionRejectedCount();
+
+ try (FileUploadDownloadClient fileUploadDownloadClient = new
FileUploadDownloadClient()) {
+ assertEquals(
+
fileUploadDownloadClient.uploadSegment(URI.create(getOrCreateAdminClient().getSegmentUploadUrl()),
+ segmentTarFile.getName(), segmentTarFile, getTableName(),
TableType.OFFLINE).getStatusCode(),
+ HttpStatus.SC_OK);
+ }
+
+ assertEquals(getSegmentsList().size(), 1);
+ assertEquals(getOutOfRetentionRejectedCount(), rejectedBefore);
+ }
+
+ private File createTableAndBuildSegment(String retentionDays)
+ throws Exception {
+ Schema schema = createSchema();
+ addSchema(schema);
+ TableConfig offlineTableConfig = createOfflineTableConfig();
+ offlineTableConfig.getValidationConfig().setRetentionTimeUnit("DAYS");
+
offlineTableConfig.getValidationConfig().setRetentionTimeValue(retentionDays);
+ waitForEVToDisappear(offlineTableConfig.getTableName());
+ addTableConfig(offlineTableConfig);
+
+ ClusterIntegrationTestUtils.buildSegmentFromAvro(getAllAvroFiles().get(0),
offlineTableConfig, schema,
+ "_retention", _segmentDir, _tarDir);
+ File[] tarFiles = _tarDir.listFiles();
+ assertNotNull(tarFiles);
+ assertEquals(tarFiles.length, 1);
+ return tarFiles[0];
+ }
+
+ private long getOutOfRetentionRejectedCount() {
+ PinotMeter meter = _controllerStarter.getControllerMetrics()
+
.getMeteredValue(ControllerMeter.OUT_OF_RETENTION_SEGMENT_UPLOAD_REJECTED);
+ return meter != null ? meter.count() : 0L;
+ }
+
protected TableConfig createOfflineTableConfigWithConsistentPush() {
TableConfig offlineTableConfig = createOfflineTableConfig();
IngestionConfig ingestionConfig = new IngestionConfig();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
