private static Pair<String, SnapshotIndex> verifyLedger(String storeName,
      Optional<Pair<String, SnapshotIndex>> startingSnapshot,
      boolean hostAffinity, boolean verifySideInputOffsetsUploaded, boolean verifySideInputOffsetsRestored) {
    Path ledgerLocation = Paths.get(BLOB_STORE_LEDGER_DIR);
    try {
      File filesAddedLedger = Paths.get(ledgerLocation.toString(), TestBlobStoreManager.LEDGER_FILES_ADDED).toFile();
      Set<String> filesAdded = Files.lines(filesAddedLedger.toPath()).filter(l -> l.contains(storeName)).collect(Collectors.toSet());
      File filesReadLedger = Paths.get(ledgerLocation.toString(), TestBlobStoreManager.LEDGER_FILES_READ).toFile();
      Set<String> filesRead = Files.lines(filesReadLedger.toPath()).filter(l -> l.contains(storeName)).collect(Collectors.toSet());
      File filesDeletedLedger = Paths.get(ledgerLocation.toString(), TestBlobStoreManager.LEDGER_FILES_DELETED).toFile();
      Set<String> filesDeleted = Files.lines(filesDeletedLedger.toPath()).filter(l -> l.contains(storeName)).collect(Collectors.toSet());
      File filesTTLUpdatedLedger = Paths.get(ledgerLocation.toString(), TestBlobStoreManager.LEDGER_FILES_TTL_UPDATED).toFile();
      Set<String> filesTTLUpdated = Files.lines(filesTTLUpdatedLedger.toPath()).filter(l -> l.contains(storeName)).collect(Collectors.toSet());

      // 1. test that files read = files present in last snapshot *at run start* + snapshot file itself + previous snapshot files
      if (startingSnapshot.isPresent() && !hostAffinity) { // no restore if host affinity (local state already present)
        Set<String> filesPresentInStartingSnapshot = startingSnapshot.get().getRight()
            .getDirIndex().getFilesPresent().stream()
            .map(fi -> fi.getBlobs().get(0).getBlobId()).collect(Collectors.toSet());
        Set<String> filesToRestore = new HashSet<>();
        filesToRestore.add(startingSnapshot.get().getLeft());
        filesToRestore.addAll(filesPresentInStartingSnapshot);
        // assert that all files to restore in starting snapshot + starting snapshot itself are present in files read
        assertTrue(Sets.difference(filesToRestore, filesRead).isEmpty());
        // assert that the remaining read files are all snapshot indexes (for post commit cleanup)
        assertTrue(Sets.difference(filesRead, filesToRestore).stream().allMatch(s -> s.contains(Metadata.SNAPSHOT_INDEX_PAYLOAD_PATH)));
      }

      // read files added again as ordered list, not set, to get last file added
      List<String> filesAddedLines = Files.readAllLines(filesAddedLedger.toPath()).stream().filter(l -> l.contains(storeName)).collect(Collectors.toList());
      String lastFileAdded = filesAddedLines.get(filesAddedLines.size() - 1); // get last line.
      assertTrue(lastFileAdded.contains(Metadata.SNAPSHOT_INDEX_PAYLOAD_PATH)); // assert is a snapshot
      SnapshotIndex lastSnapshotIndex = new SnapshotIndexSerde().fromBytes(Files.readAllBytes(Paths.get(lastFileAdded)));
      Set<String> filesPresentInLastSnapshot = lastSnapshotIndex.getDirIndex().getFilesPresent().stream()
          .map(fi -> fi.getBlobs().get(0).getBlobId()).collect(Collectors.toSet());

      // 2. test that all added files were ttl reset
      assertEquals(filesAdded, filesTTLUpdated);

      // 3. test that files deleted = files added - files present in last snapshot + snapshot file itself
      // i.e., net remaining files (files added - files deleted) = files present in last snapshot + snapshot file itself.
      assertEquals(Sets.difference(filesAdded, filesDeleted),
          Sets.union(filesPresentInLastSnapshot, Collections.singleton(lastFileAdded)));

      // 4. test that the files restored/added for side input stores contains side input offsets file
      if (verifySideInputOffsetsUploaded) {
        assertTrue(filesAdded.stream().anyMatch(f -> f.contains(StorageManagerUtil.SIDE_INPUT_OFFSET_FILE_NAME_LEGACY)));
      }

      if (!hostAffinity && verifySideInputOffsetsRestored) { // only read / restored if no host affinity
        assertTrue(filesRead.stream().anyMatch(f -> f.contains(StorageManagerUtil.SIDE_INPUT_OFFSET_FILE_NAME_LEGACY)));
      }

      return Pair.of(lastFileAdded, lastSnapshotIndex);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
Verifies the ledger for TestBlobStoreManager.
@param startingSnapshot snapshot file name and files present in snapshot at the beginning of verification (from previous run), if any.
@return Pair file for latest snapshot at time of verification
samza-test/src/test/java/org/apache/samza/storage/kv/BlobStoreStateBackendIntegrationTest.java