public static StatementExecutorResponse execute(
      final ConfiguredStatement<TerminateQuery> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final TerminateQuery terminateQuery = statement.getStatement();
    // do default behaviour for TERMINATE ALL
    if (!terminateQuery.getQueryId().isPresent()) {
      return StatementExecutorResponse.notHandled();
    }
    final QueryId queryId = terminateQuery.getQueryId().get();
    final RemoteHostExecutor remoteHostExecutor = RemoteHostExecutor.create(
        statement,
        sessionProperties,
        executionContext,
        serviceContext.getKsqlClient()
    );

    if (executionContext.getPersistentQuery(queryId).isPresent()
        || statement.getUnMaskedStatementText().equals(
            TerminateCluster.TERMINATE_CLUSTER_STATEMENT_TEXT)) {
      // do default behaviour for terminating persistent queries
      return StatementExecutorResponse.notHandled();
    } else {
      // Check are we running this push query locally, if yes then terminate, otherwise
      // propagate terminate query to other nodes
      if (executionContext.getQuery(queryId).isPresent()) {
        executionContext.getQuery(queryId).get().close();
      } else {
        final boolean wasTerminatedRemotely = remoteHostExecutor.fetchAllRemoteResults().getLeft()
            .values()
            .stream()
            .map(TerminateQueryEntity.class::cast)
            .map(TerminateQueryEntity::getWasTerminated)
            .anyMatch(b -> b.equals(true));
        if (!wasTerminatedRemotely) {
          throw new KsqlException(String.format(
              "Failed to terminate query with query ID: '%s'",
              queryId));
        }
      }
      return StatementExecutorResponse.handled(Optional.of(
          new TerminateQueryEntity(statement.getMaskedStatementText(), queryId.toString(), true)
      ));
    }
  }
@Test
  public void shouldDefaultToDistributorForPersistentQuery() {
    // Given:
    final ConfiguredStatement<?> terminatePersistent = engine.configure("TERMINATE PERSISTENT_QUERY;");
    final PersistentQueryMetadata persistentQueryMetadata = givenPersistentQuery("PERSISTENT_QUERY", RUNNING_QUERY_STATE);
    final QueryId persistentQueryId= persistentQueryMetadata.getQueryId();

    final KsqlEngine engine = mock(KsqlEngine.class);
    when(engine.getPersistentQuery(persistentQueryId)).thenReturn(Optional.of(persistentQueryMetadata));

    // When:
    final Optional<KsqlEntity> ksqlEntity = CustomExecutors.TERMINATE_QUERY.execute(
        terminatePersistent,
        mock(SessionProperties.class),
        engine,
        this.engine.getServiceContext()
    ).getEntity();

    // Then:
    assertThat(ksqlEntity, is(Optional.empty()));
  }