public void cleanupOrphanedInternalTopics(
      final ServiceContext serviceContext,
      final Set<String> queryApplicationIds
  ) {
    final KafkaTopicClient topicClient = serviceContext.getTopicClient();
    final Set<String> topicNames;
    try {
      topicNames = topicClient.listTopicNames();
    } catch (KafkaResponseGetFailedException e) {
      LOG.error("Couldn't fetch topic names", e);
      return;
    }
    // Find any transient query topics
    final Set<String> orphanedQueryApplicationIds = topicNames.stream()
        .map(topicName -> queryApplicationIds.stream().filter(topicName::startsWith).findFirst())
        .filter(Optional::isPresent)
        .map(Optional::get)
        .collect(Collectors.toSet());
    for (final String queryApplicationId : orphanedQueryApplicationIds) {
      cleanupService.addCleanupTask(
          new QueryCleanupService.QueryCleanupTask(
              serviceContext,
              queryApplicationId,
              Optional.empty(),
              true,
              ksqlConfig.getKsqlStreamConfigProps()
                  .getOrDefault(
                      StreamsConfig.STATE_DIR_CONFIG,
                      StreamsConfig.configDef().defaultValues().get(StreamsConfig.STATE_DIR_CONFIG))
                  .toString(),
              ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
              ksqlConfig.getString(KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG)));
    }
  }
@Test
  public void shouldSkip_exception() {
    // Given
    when(topicClient.listTopicNames())
        .thenThrow(new KafkaResponseGetFailedException("error!", new Exception()));

    // When
    cleaner.cleanupOrphanedInternalTopics(serviceContext, ImmutableSet.of());

    // Then
    verify(queryCleanupService, never()).addCleanupTask(any());
  }