@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 testPosDeleteRecordCount() throws Exception {
    TriggerManager manager =
        manager(
            sql.tableLoader(TABLE_NAME),
            new TriggerEvaluator.Builder().posDeleteRecordCount(3).build());
    try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
        harness(manager)) {
      testHarness.open();

      addEventAndCheckResult(
          testHarness, TableChange.builder().posDeleteRecordCount(1L).build(), 0);
      addEventAndCheckResult(
          testHarness, TableChange.builder().posDeleteRecordCount(2L).build(), 1);
      addEventAndCheckResult(
          testHarness, TableChange.builder().posDeleteRecordCount(5L).build(), 2);

      // No trigger in this case
      addEventAndCheckResult(
          testHarness, TableChange.builder().posDeleteRecordCount(1L).build(), 2);

      addEventAndCheckResult(
          testHarness, TableChange.builder().posDeleteRecordCount(2L).build(), 3);
    }
  }