public void close(final boolean closeQueries) {
    primaryContext.getQueryRegistry().close(closeQueries);

    try {
      cleanupService.stopAsync().awaitTerminated(
          this.primaryContext.getKsqlConfig()
              .getLong(KsqlConfig.KSQL_QUERY_CLEANUP_SHUTDOWN_TIMEOUT_MS),
          TimeUnit.MILLISECONDS);
    } catch (final TimeoutException e) {
      log.warn("Timed out while closing cleanup service. "
              + "External resources for the following applications may be orphaned: {}",
          cleanupService.pendingApplicationIds()
      );
    }

    engineMetrics.close();
    aggregateMetricsCollector.shutdown();
  }
@Test
  public void shouldCleanUpInternalTopicsOnEngineCloseForTransientQueriesSharedRuntimes() {
    // Given:
    setupKsqlEngineWithSharedRuntimeEnabled();

    final QueryMetadata query = KsqlEngineTestUtil.executeQuery(
        serviceContext,
        ksqlEngine,
        "select * from test1 EMIT CHANGES;",
        ksqlConfig,
        Collections.emptyMap()
    );

    query.start();

    // When:
    ksqlEngine.close();
    isKsqlEngineClosed = true;

    // Then:
    verify(topicClient, times(2)).deleteInternalTopics(query.getQueryApplicationId());
  }