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 shouldThrowOnDenyListValidatorWhenTerminateCluster() {
    final Map<String, Object> terminateStreamProperties =
        ImmutableMap.of(DELETE_TOPIC_LIST_PROP, Collections.singletonList("Foo"));

    // Given:
    doThrow(new KsqlException("deny override")).when(denyListPropertyValidator).validateAll(
        terminateStreamProperties
    );

    // When:
    final EndpointResponse response = ksqlResource.terminateCluster(
        securityContext,
        VALID_TERMINATE_REQUEST
    );

    // Then:
    verify(denyListPropertyValidator).validateAll(terminateStreamProperties);
    assertThat(response.getStatus(), equalTo(INTERNAL_SERVER_ERROR.code()));
    assertThat(response.getEntity(), instanceOf(KsqlStatementErrorMessage.class));
    assertThat(((KsqlStatementErrorMessage) response.getEntity()).getMessage(),
        containsString("deny override"));
  }