@Test
  @SuppressWarnings("ResultOfMethodCallIgnored")
  public void testDeleteLoggedStoreOnNoCheckpoints() {
    // reset the mock to reset the stubs in setup method
    reset(this.checkpointManager);
    // redo stubbing to return null checkpoints
    when(this.checkpointManager.readLastCheckpoint(any())).thenReturn(null);
    // create store under logged stores to demonstrate deletion
    final File storeFile = new File(DEFAULT_LOGGED_STORE_BASE_DIR.getPath() + File.separator + STORE_NAME);
    // add contents to store
    final File storeFilePartition = new File(DEFAULT_LOGGED_STORE_BASE_DIR.getPath() + File.separator + STORE_NAME + File.separator + "Partition_0");
    storeFilePartition.deleteOnExit();
    storeFile.deleteOnExit();
    try {
      storeFile.mkdirs();
      storeFilePartition.createNewFile();
      Assert.assertTrue("Assert that stores are present prior to the test.", storeFile.exists());
      Assert.assertTrue("Assert that store files are present prior to the test.", storeFilePartition.exists());
      this.containerStorageManager.start();
      this.containerStorageManager.shutdown();
      Assert.assertFalse("Assert that stores are deleted after the test.", storeFile.exists());
      Assert.assertFalse("Assert that store files are deleted after the test.", storeFilePartition.exists());
    } catch (Exception e) {
      System.out.printf("File %s could not be created.", storeFile);
      Assert.fail();
    } finally {
      storeFilePartition.delete();
      storeFile.delete();
    }
  }
This test will attempt to verify if logged stores are deleted if the input checkpoints are empty.
samza-core/src/test/scala/org/apache/samza/storage/TestContainerStorageManager.java