@Override
  public void close() throws Exception {
    super.close();

    if (checkpointLock != null) {
      synchronized (checkpointLock) {
        issuedInstant = null;
        isRunning = false;
      }
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("Closed File Monitoring Source for path: " + path + ".");
    }
  }
@Test
  public void testConsumingHollowInstants() throws Exception {
    // write 4 commits
    conf.setString("hoodie.parquet.small.file.limit", "0"); // invalidate the small file strategy
    for (int i = 0; i < 8; i += 2) {
      List<RowData> dataset = TestData.dataSetInsert(i + 1, i + 2);
      TestData.writeData(dataset, conf);
    }

    // we got 4 commits on the timeline: c1, c2, c3, c4
    // re-create the metadata file for c2 and c3 so that they have greater completion time than c4.
    // the completion time sequence become: c1, c4, c2, c3,
    // we will test with the same consumption sequence.
    HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf);
    List<HoodieInstant> oriInstants = metaClient.getCommitsTimeline().filterCompletedInstants().getInstants();
    assertThat(oriInstants.size(), is(4));
    List<HoodieCommitMetadata> metadataList = new ArrayList<>();
    // timeline: c1, c2.inflight, c3.inflight, c4
    for (int i = 1; i <= 2; i++) {
      HoodieInstant instant = oriInstants.get(i);
      metadataList.add(TestUtils.deleteInstantFile(metaClient, instant));
    }

    List<HoodieInstant> instants = metaClient.reloadActiveTimeline().getCommitsTimeline().filterCompletedInstants().getInstants();
    assertThat(instants.size(), is(2));

    String c2 = oriInstants.get(1).getTimestamp();
    String c3 = oriInstants.get(2).getTimestamp();
    String c4 = instants.get(1).getTimestamp();

    conf.setString(FlinkOptions.READ_START_COMMIT, FlinkOptions.START_COMMIT_EARLIEST);
    StreamReadMonitoringFunction function = TestUtils.getMonitorFunc(conf);
    try (AbstractStreamOperatorTestHarness<MergeOnReadInputSplit> harness = createHarness(function)) {
      harness.setup();
      harness.open();

      // timeline: c1, c2.inflight, c3.inflight, c4
      // -> c1
      CountDownLatch latch = new CountDownLatch(2);
      CollectingSourceContext sourceContext = new CollectingSourceContext(latch);

      runAsync(sourceContext, function);

      assertTrue(latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS), "Should finish splits generation");
      assertThat("Should produce the expected splits",
          sourceContext.getPartitionPaths(), is("par1"));
      assertTrue(sourceContext.splits.stream().noneMatch(split -> split.getInstantRange().isPresent()),
          "No instants should have range limit");
      assertTrue(sourceContext.splits.stream().anyMatch(split -> split.getLatestCommit().equals(c4)),
          "At least one input split's latest commit time should be equal to the specified instant time.");

      // reset the source context
      latch = new CountDownLatch(1);
      sourceContext.reset(latch);

      // timeline: c1, c2, c3.inflight, c4
      // c4 -> c2
      TestUtils.saveInstantAsComplete(metaClient, oriInstants.get(1), metadataList.get(0)); // complete c2
      assertTrue(latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS), "Should finish splits generation");
      assertThat("Should produce the expected splits",
          sourceContext.getPartitionPaths(), is("par1"));
      assertTrue(sourceContext.splits.stream().allMatch(split -> split.getInstantRange().isPresent()),
          "All instants should have range limit");
      assertTrue(sourceContext.splits.stream().allMatch(split -> isPointInstantRange(split.getInstantRange().get(), c2)),
          "All the splits should have point instant range");
      assertTrue(sourceContext.splits.stream().anyMatch(split -> split.getLatestCommit().equals(c2)),
          "At least one input split's latest commit time should be equal to the specified instant time.");

      // reset the source context
      latch = new CountDownLatch(1);
      sourceContext.reset(latch);

      // timeline: c1, c2, c3, c4
      // c4 -> c3
      TestUtils.saveInstantAsComplete(metaClient, oriInstants.get(2), metadataList.get(1)); // complete c3
      assertTrue(latch.await(WAIT_TIME_MILLIS, TimeUnit.MILLISECONDS), "Should finish splits generation");
      assertThat("Should produce the expected splits",
          sourceContext.getPartitionPaths(), is("par1"));
      assertTrue(sourceContext.splits.stream().allMatch(split -> split.getInstantRange().isPresent()),
          "All instants should have range limit");
      assertTrue(sourceContext.splits.stream().allMatch(split -> isPointInstantRange(split.getInstantRange().get(), c3)),
          "All the splits should have point instant range");
      assertTrue(sourceContext.splits.stream().anyMatch(split -> split.getLatestCommit().equals(c3)),
          "At least one input split's latest commit time should be equal to the specified instant time.");

      // Stop the stream task.
      function.close();
    }
  }