public void cleanupLeakedQueries(final List<PersistentQueryMetadata> persistentQueries) {
    final Set<String> stateStoreNames =
        persistentQueries
            .stream()
            .flatMap(s -> {
              final List<String> doNotDelete = new ArrayList<>(
                  Collections.singletonList(s.getQueryApplicationId()));
              if (s instanceof BinPackedPersistentQueryMetadataImpl) {
                doNotDelete.add(s.getQueryApplicationId()
                    + "/__"
                    + s.getQueryId().toString()
                    + "__");
              }
              return doNotDelete.stream();
            })
            .collect(Collectors.toSet());

    final String[] stateDirFileNames = new File(stateDir).list();
    if (stateDirFileNames == null) {
      LOG.info("No state stores to clean up");
    } else {
      final Set<String> allStateStores = Arrays.stream(stateDirFileNames)
          .flatMap(f -> {
            final String[] fileNames =  new File(stateDir + "/" + f).list();
            if (null == fileNames) {
              return Stream.of(f);
            } else if (Arrays.stream(fileNames).anyMatch(t -> t.matches("__*__"))) {
              return Arrays.stream(fileNames)
                  .filter(t -> t.matches("__*__"))
                  .map(s -> f + "/" + s);
            } else {
              return Stream.of(f);
            }
          })
          .collect(Collectors.toSet());
      allStateStores.removeAll(stateStoreNames);
      allStateStores.forEach((storeName) -> queryCleanupService.addCleanupTask(
          new QueryCleanupService.QueryCleanupTask(
            serviceContext,
            storeName.split("/")[0],
            1 <  storeName.split("__").length
                ? Optional.of(storeName.split("__")[1])
                : Optional.empty(),
            false,
            stateDir,
            ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
            ksqlConfig.getString(KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG))));
    }
  }
@Test
  public void shouldNotDeleteSharedRuntimesWhenTheyHaveAQuery() {
    // Given:
    when(binPackedPersistentQueryMetadata.getQueryApplicationId()).thenReturn("testQueryID");
    File fakeStateStore = new File(tempFile.getAbsolutePath() + "testQueryID");
    if (!fakeStateStore.exists()) {
      assertTrue(fakeStateStore.mkdirs());
    }
    // When:
    cleanup.cleanupLeakedQueries(ImmutableList.of(binPackedPersistentQueryMetadata));
    awaitCleanupComplete();

    // Then:
    assertTrue(fakeStateStore.exists());
    assertTrue(tempFile.exists());
  }