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 shouldHandleTerminateRequestCorrectly() {
    // When:
    final EndpointResponse response = ksqlResource.terminateCluster(
        securityContext,
        VALID_TERMINATE_REQUEST
    );

    // Then:
    assertThat(response.getStatus(), equalTo(200));
    assertThat(response.getEntity(), instanceOf(KsqlEntityList.class));
    assertThat(((KsqlEntityList) response.getEntity()).size(), equalTo(1));
    assertThat(((KsqlEntityList) response.getEntity()).get(0),
        instanceOf(CommandStatusEntity.class));
    final CommandStatusEntity commandStatusEntity =
        (CommandStatusEntity) ((KsqlEntityList) response.getEntity()).get(0);
    assertThat(commandStatusEntity.getCommandStatus().getStatus(),
        equalTo(CommandStatus.Status.QUEUED));
    verify(transactionalProducer, times(1)).initTransactions();
    verify(commandStore).enqueueCommand(
        any(),
        argThat(is(commandWithStatement(TerminateCluster.TERMINATE_CLUSTER_STATEMENT_TEXT))),
        any()
    );
  }