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 shouldFailIfCannotWriteTerminateCommand() {
    // Given:
    when(commandStore.enqueueCommand(any(), any(), any(Producer.class)))
        .thenThrow(new KsqlException(""));

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

    // Then:
    assertThat(response.getStatus(), equalTo(500));
    assertThat(response.getEntity().toString(),
        CoreMatchers
            .startsWith("Could not write the statement 'TERMINATE CLUSTER;' into the command "));
    assertThat(response.getEntity(), instanceOf(KsqlStatementErrorMessage.class));
    final KsqlStatementErrorMessage entity = (KsqlStatementErrorMessage) response.getEntity();
    assertThat(entity.getStatementText(), containsString("TERMINATE CLUSTER"));
  }