public void terminateCluster(final List<String> deleteTopicPatterns) {
    terminatePersistentQueries();
    deleteSinkTopics(deleteTopicPatterns);
    deleteTopics(managedTopics);
    ksqlEngine.close();
  }
@Test
  public void shouldCleanUpSchemasForTopicListWithPattern() throws Exception {
    // Given:
    givenTopicsExistInKafka("K_Fo", "K_Foo", "K_Fooo", "NotMatched");
    givenSinkTopicsExistInMetastore(FormatFactory.AVRO, "K_Fo", "K_Foo", "K_Fooo", "NotMatched");
    givenSchemasForTopicsExistInSchemaRegistry("K_Fo", "K_Foo", "K_Fooo", "NotMatched");

    // When:
    clusterTerminator.terminateCluster(ImmutableList.of("K_Fo.*"));

    // Then:
    verifySchemaDeletedForTopics("K_Foo", "K_Fooo", "K_Fo");
    verifySchemaNotDeletedForTopic("NotMatched");
  }