@Override
  public boolean isPurgeable(String tableNameWithType, SegmentZKMetadata segmentZKMetadata) {
    long endTimeMs = segmentZKMetadata.getEndTimeMs();

    // Check that the end time is between 1971 and 2071
    if (!TimeUtils.timeValueInValidRange(endTimeMs)) {
      LOGGER.warn("Segment: {} of table: {} has invalid end time in millis: {}", segmentZKMetadata.getSegmentName(),
          tableNameWithType, endTimeMs);
      return false;
    }

    return System.currentTimeMillis() - endTimeMs > _retentionMs;
  }
@Test
  public void testTimeRetention() {
    String tableNameWithType = "myTable_OFFLINE";
    TimeRetentionStrategy retentionStrategy = new TimeRetentionStrategy(TimeUnit.DAYS, 30L);

    SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata("mySegment");

    // Without setting time unit or end time, should not throw exception
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));
    segmentZKMetadata.setTimeUnit(TimeUnit.DAYS);
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));

    // Set end time to Jan 2nd, 1970 (not purgeable due to bogus timestamp)
    segmentZKMetadata.setEndTime(1L);
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));

    // Set end time to today
    long today = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
    segmentZKMetadata.setEndTime(today);
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));

    // Set end time to two weeks ago
    segmentZKMetadata.setEndTime(today - 14);
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));

    // Set end time to two months ago (purgeable due to being past the retention period)
    segmentZKMetadata.setEndTime(today - 60);
    assertTrue(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));

    // Set end time to 200 years in the future (not purgeable due to bogus timestamp)
    segmentZKMetadata.setEndTime(today + (365 * 200));
    assertFalse(retentionStrategy.isPurgeable(tableNameWithType, segmentZKMetadata));
  }