public TimeBoundaryManager(TableConfig tableConfig, ZkHelixPropertyStore<ZNRecord> propertyStore,
      BrokerMetrics brokerMetrics) {
    Preconditions.checkState(tableConfig.getTableType() == TableType.OFFLINE,
        "Cannot construct TimeBoundaryManager for real-time table: %s", tableConfig.getTableName());
    _offlineTableName = tableConfig.getTableName();
    _propertyStore = propertyStore;
    _brokerMetrics = brokerMetrics;
    _segmentZKMetadataPathPrefix = ZKMetadataProvider.constructPropertyStorePathForResource(_offlineTableName) + "/";

    Schema schema = ZKMetadataProvider.getTableSchema(_propertyStore, _offlineTableName);
    Preconditions.checkState(schema != null, "Failed to find schema for table: %s", _offlineTableName);
    _timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
    Preconditions.checkNotNull(_timeColumn, "Time column must be configured in table config for table: %s",
        _offlineTableName);
    DateTimeFieldSpec dateTimeSpec = schema.getSpecForTimeColumn(_timeColumn);
    Preconditions.checkNotNull(dateTimeSpec, "Field spec must be specified in schema for time column: %s of table: %s",
        _timeColumn, _offlineTableName);
    _timeFormatSpec = dateTimeSpec.getFormatSpec();
    Preconditions.checkNotNull(_timeFormatSpec.getColumnUnit(),
        "Time unit must be configured in the field spec for time column: %s of table: %s", _timeColumn,
        _offlineTableName);

    // For HOURLY table with time unit other than DAYS, use (maxEndTime - 1 HOUR) as the time boundary; otherwise, use
    // (maxEndTime - 1 DAY)
    boolean isHourlyTable = CommonConstants.Table.PUSH_FREQUENCY_HOURLY.equalsIgnoreCase(
        IngestionConfigUtils.getBatchSegmentIngestionFrequency(tableConfig))
        && _timeFormatSpec.getColumnUnit() != TimeUnit.DAYS;
    _timeOffsetMs = isHourlyTable ? TimeUnit.HOURS.toMillis(1) : TimeUnit.DAYS.toMillis(1);

    LOGGER.info("Constructed TimeBoundaryManager with timeColumn: {}, timeFormat: {}, isHourlyTable: {} for table: {}",
        _timeColumn, dateTimeSpec.getFormat(), isHourlyTable, _offlineTableName);
  }
@Test
  public void testTimeBoundaryManager() {
    for (TimeUnit timeUnit : TimeUnit.values()) {
      // Test DAILY push table, with timeFieldSpec
      String rawTableName = "testTable_" + timeUnit + "_DAILY";
      TableConfig tableConfig = getTableConfig(rawTableName, "DAILY");
      setSchemaTimeFieldSpec(rawTableName, timeUnit);
      testDailyPushTable(rawTableName, tableConfig, timeUnit);

      // Test HOURLY push table, with timeFieldSpec
      rawTableName = "testTable_" + timeUnit + "_HOURLY";
      tableConfig = getTableConfig(rawTableName, "HOURLY");
      setSchemaTimeFieldSpec(rawTableName, timeUnit);
      testHourlyPushTable(rawTableName, tableConfig, timeUnit);

      // Test DAILY push table with dateTimeFieldSpec
      rawTableName = "testTableDateTime_" + timeUnit + "_DAILY";
      tableConfig = getTableConfig(rawTableName, "DAILY");
      setSchemaDateTimeFieldSpec(rawTableName, timeUnit);
      testDailyPushTable(rawTableName, tableConfig, timeUnit);

      // Test HOURLY push table
      rawTableName = "testTableDateTime_" + timeUnit + "_HOURLY";
      tableConfig = getTableConfig(rawTableName, "HOURLY");
      setSchemaDateTimeFieldSpec(rawTableName, timeUnit);
      testHourlyPushTable(rawTableName, tableConfig, timeUnit);
    }
  }