public EndpointResponse terminateCluster(
      final KsqlSecurityContext securityContext,
      final ClusterTerminateRequest request
  ) {
    LOG.info("Received: " + request);

    throwIfNotConfigured();

    ensureValidPatterns(request.getDeleteTopicList());
    try {
      final Map<String, Object> streamsProperties = request.getStreamsProperties();
      denyListPropertyValidator.validateAll(streamsProperties);

      final KsqlEntityList entities = handler.execute(
          securityContext,
          TERMINATE_CLUSTER,
          new SessionProperties(
              streamsProperties,
              localHost,
              localUrl,
              false
          )
      );
      return EndpointResponse.ok(entities);
    } catch (final Exception e) {
      return Errors.serverErrorForStatement(
          e, TerminateCluster.TERMINATE_CLUSTER_STATEMENT_TEXT, new KsqlEntityList());
    }
  }
@Test
  public void shouldThrowOnHandleTerminateIfNotConfigured() {
    // Given:
    ksqlResource = new KsqlResource(
        ksqlEngine,
        commandRunner,
        DISTRIBUTED_COMMAND_RESPONSE_TIMEOUT,
        activenessRegistrar,
        (ec, sc) -> InjectorChain.of(
            schemaInjectorFactory.apply(sc),
            topicInjectorFactory.apply(ec),
            new TopicDeleteInjector(ec, sc)),
        Optional.of(authorizationValidator),
        errorsHandler,
        denyListPropertyValidator,
        commandRunnerWarning
    );

    // When:
    final KsqlRestException e = assertThrows(
        KsqlRestException.class,
        () -> ksqlResource.terminateCluster(
            securityContext,
            new ClusterTerminateRequest(ImmutableList.of(""))
        )
    );

    // Then:
    assertThat(e, exceptionStatusCode(CoreMatchers.is(SERVICE_UNAVAILABLE.code())));
    assertThat(e, exceptionErrorMessage(errorMessage(Matchers.is("Server initializing"))));
  }