public synchronized void reload(long checkpointId) {
    this.accCkp += 1;
    if (this.accCkp > 1) {
      // do not clean the new file assignment state for the first checkpoint,
      // this #reload calling is triggered by checkpoint success event, the coordinator
      // also relies on the checkpoint success event to commit the inflight instant,
      // and very possibly this component receives the notification before the coordinator,
      // if we do the cleaning, the records processed within the time range:
      // (start checkpoint, checkpoint success(and instant committed))
      // would be assigned to a fresh new data bucket which is not the right behavior.
      this.newFileAssignStates.clear();
      this.accCkp = 0;
    }
    this.smallFileAssignMap.clear();
    this.writeProfile.reload(checkpointId);
  }
@Test
  public void testWriteProfileMetadataCache() throws Exception {
    WriteProfile writeProfile = new WriteProfile(writeConfig, context);
    assertTrue(writeProfile.getMetadataCache().isEmpty(), "Empty table should no have any instant metadata");

    // write 3 instants of data
    for (int i = 0; i < 3; i++) {
      TestData.writeData(TestData.DATA_SET_INSERT, conf);
    }
    // the record profile triggers the metadata loading
    writeProfile.reload(1);
    assertThat("Metadata cache should have same number entries as timeline instants",
        writeProfile.getMetadataCache().size(), is(3));

    writeProfile.getSmallFiles("par1");
    assertThat("The metadata should be reused",
        writeProfile.getMetadataCache().size(), is(3));
  }