@Test
  public void TestIncBackupDeleteTable() throws Exception {
    try (BackupSystemTable systemTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
      // The test starts with some data, and no bulk loaded rows.
      int expectedRowCount = NB_ROWS_IN_BATCH;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertTrue(systemTable.readBulkloadRows(List.of(table1)).isEmpty());

      // Bulk loads aren't tracked if the table isn't backed up yet
      performBulkLoad("bulk1");
      expectedRowCount += ROWS_IN_BULK_LOAD;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(0, systemTable.readBulkloadRows(List.of(table1)).size());

      // Create a backup, bulk loads are now being tracked
      String backup1 = backupTables(BackupType.FULL, List.of(table1), BACKUP_ROOT_DIR);
      assertTrue(checkSucceeded(backup1));
      performBulkLoad("bulk2");
      expectedRowCount += ROWS_IN_BULK_LOAD;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(1, systemTable.readBulkloadRows(List.of(table1)).size());

      // Truncating or deleting a table clears the tracked bulk loads (and all rows)
      TEST_UTIL.truncateTable(table1).close();
      expectedRowCount = 0;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(0, systemTable.readBulkloadRows(List.of(table1)).size());

      // Creating a full backup clears the bulk loads (since they are captured in the snapshot)
      performBulkLoad("bulk3");
      expectedRowCount = ROWS_IN_BULK_LOAD;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(1, systemTable.readBulkloadRows(List.of(table1)).size());
      String backup2 = backupTables(BackupType.FULL, List.of(table1), BACKUP_ROOT_DIR);
      assertTrue(checkSucceeded(backup2));
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(0, systemTable.readBulkloadRows(List.of(table1)).size());

      // Creating an incremental backup clears the bulk loads
      performBulkLoad("bulk4");
      performBulkLoad("bulk5");
      performBulkLoad("bulk6");
      expectedRowCount += 3 * ROWS_IN_BULK_LOAD;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(3, systemTable.readBulkloadRows(List.of(table1)).size());
      String backup3 = backupTables(BackupType.INCREMENTAL, List.of(table1), BACKUP_ROOT_DIR);
      assertTrue(checkSucceeded(backup3));
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      assertEquals(0, systemTable.readBulkloadRows(List.of(table1)).size());
      int rowCountAfterBackup3 = expectedRowCount;

      // Doing another bulk load, to check that this data will disappear after a restore operation
      performBulkLoad("bulk7");
      expectedRowCount += ROWS_IN_BULK_LOAD;
      assertEquals(expectedRowCount, TEST_UTIL.countRows(table1));
      List<BulkLoad> bulkloadsTemp = systemTable.readBulkloadRows(List.of(table1));
      assertEquals(1, bulkloadsTemp.size());
      BulkLoad bulk7 = bulkloadsTemp.get(0);

      // Doing a restore. Overwriting the table implies clearing the bulk loads,
      // but the loading of restored data involves loading bulk data, we expect 2 bulk loads
      // associated with backup 3 (loading of full backup, loading of incremental backup).
      BackupAdmin client = getBackupAdmin();
      client.restore(BackupUtils.createRestoreRequest(BACKUP_ROOT_DIR, backup3, false,
        new TableName[] { table1 }, new TableName[] { table1 }, true));
      assertEquals(rowCountAfterBackup3, TEST_UTIL.countRows(table1));
      List<BulkLoad> bulkLoads = systemTable.readBulkloadRows(List.of(table1));
      assertEquals(2, bulkLoads.size());
      assertFalse(bulkLoads.contains(bulk7));

      // Check that we have data of all expected bulk loads
      try (Table restoredTable = TEST_UTIL.getConnection().getTable(table1)) {
        assertFalse(containsRowWithKey(restoredTable, "bulk1"));
        assertFalse(containsRowWithKey(restoredTable, "bulk2"));
        assertTrue(containsRowWithKey(restoredTable, "bulk3"));
        assertTrue(containsRowWithKey(restoredTable, "bulk4"));
        assertTrue(containsRowWithKey(restoredTable, "bulk5"));
        assertTrue(containsRowWithKey(restoredTable, "bulk6"));
        assertFalse(containsRowWithKey(restoredTable, "bulk7"));
      }
    }
  }
This test checks whether backups properly track & manage bulk files loads.
hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackupWithBulkLoad.java