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 shouldHardDeleteSchemaOnEngineCloseForTransientQueriesSharedRuntimes() throws IOException, RestClientException {
    // Given:
    setupKsqlEngineWithSharedRuntimeEnabled();
    final QueryMetadata query = KsqlEngineTestUtil.executeQuery(
        serviceContext,
        ksqlEngine,
        "select * from test1 EMIT CHANGES;",
        ksqlConfig, Collections.emptyMap()
    );
    final String internalTopic1Val = KsqlConstants.getSRSubject(
        query.getQueryApplicationId() + "-subject1" + KsqlConstants.STREAMS_CHANGELOG_TOPIC_SUFFIX, false);
    final String internalTopic2Val = KsqlConstants.getSRSubject(
        query.getQueryApplicationId() + "-subject3" + KsqlConstants.STREAMS_REPARTITION_TOPIC_SUFFIX, false);
    final String internalTopic1Key = KsqlConstants.getSRSubject(
        query.getQueryApplicationId() + "-subject1" + KsqlConstants.STREAMS_CHANGELOG_TOPIC_SUFFIX, true);
    final String internalTopic2Key = KsqlConstants.getSRSubject(
        query.getQueryApplicationId() + "-subject3" + KsqlConstants.STREAMS_REPARTITION_TOPIC_SUFFIX, true);

    when(schemaRegistryClient.getAllSubjects()).thenReturn(
        Arrays.asList(
            internalTopic1Val,
            internalTopic1Key,
            "subject2",
            internalTopic2Val,
            internalTopic2Key));

    query.start();

    // When:
    query.close();

    // Then:
    awaitCleanupComplete();
    verify(schemaRegistryClient, times(4)).deleteSubject(any());
    verify(schemaRegistryClient).deleteSubject(internalTopic1Val, true);
    verify(schemaRegistryClient).deleteSubject(internalTopic2Val, true);
    verify(schemaRegistryClient).deleteSubject(internalTopic1Key, true);
    verify(schemaRegistryClient).deleteSubject(internalTopic2Key, true);
    verify(schemaRegistryClient, never()).deleteSubject("subject2");
  }