@Override
  public void open(Configuration parameters) throws Exception {
    this.rateLimiterTriggeredCounter =
        getRuntimeContext()
            .getMetricGroup()
            .addGroup(
                TableMaintenanceMetrics.GROUP_KEY, TableMaintenanceMetrics.GROUP_VALUE_DEFAULT)
            .counter(TableMaintenanceMetrics.RATE_LIMITER_TRIGGERED);
    this.concurrentRunThrottledCounter =
        getRuntimeContext()
            .getMetricGroup()
            .addGroup(
                TableMaintenanceMetrics.GROUP_KEY, TableMaintenanceMetrics.GROUP_VALUE_DEFAULT)
            .counter(TableMaintenanceMetrics.CONCURRENT_RUN_THROTTLED);
    this.nothingToTriggerCounter =
        getRuntimeContext()
            .getMetricGroup()
            .addGroup(
                TableMaintenanceMetrics.GROUP_KEY, TableMaintenanceMetrics.GROUP_VALUE_DEFAULT)
            .counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
    this.triggerCounters =
        taskNames.stream()
            .map(
                name ->
                    getRuntimeContext()
                        .getMetricGroup()
                        .addGroup(TableMaintenanceMetrics.GROUP_KEY, name)
                        .counter(TableMaintenanceMetrics.TRIGGERED))
            .collect(Collectors.toList());

    this.nextEvaluationTimeState =
        getRuntimeContext()
            .getState(new ValueStateDescriptor<>("triggerManagerNextTriggerTime", Types.LONG));
    this.accumulatedChangesState =
        getRuntimeContext()
            .getListState(
                new ListStateDescriptor<>(
                    "triggerManagerAccumulatedChange", TypeInformation.of(TableChange.class)));
    this.lastTriggerTimesState =
        getRuntimeContext()
            .getListState(new ListStateDescriptor<>("triggerManagerLastTriggerTime", Types.LONG));

    tableLoader.open();
  }
@Test
  void testMinFireDelay() throws Exception {
    TableLoader tableLoader = sql.tableLoader(TABLE_NAME);
    TriggerManager manager = manager(tableLoader, DELAY, 1);
    try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
        harness(manager)) {
      testHarness.open();

      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
      long currentTime = testHarness.getProcessingTime();

      // No new fire yet
      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);

      // Check that the trigger fired after the delay
      testHarness.setProcessingTime(currentTime + DELAY);
      assertThat(testHarness.extractOutputValues()).hasSize(2);
    }
  }