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 shouldCleanUpTransientConsumerGroupsOnCloseSharedRuntimes() {
    // Given:
    setupKsqlEngineWithSharedRuntimeEnabled();
    final QueryMetadata query = KsqlEngineTestUtil.executeQuery(
        serviceContext,
        ksqlEngine,
        "select * from test1 EMIT CHANGES;",
        ksqlConfig,
        Collections.emptyMap()
    );

    query.start();

    // When:
    query.close();

    // Then:
    awaitCleanupComplete();
    final Set<String> deletedConsumerGroups = (
        (FakeKafkaConsumerGroupClient) serviceContext.getConsumerGroupClient()
    ).getDeletedConsumerGroups();

    assertThat(
        Iterables.getOnlyElement(deletedConsumerGroups),
        containsString("_confluent-ksql-default_transient_"));
  }