public EndpointResponse handleKsqlStatements(
      final KsqlSecurityContext securityContext,
      final KsqlRequest request
  ) {
    // CHECKSTYLE_RULES.ON: JavaNCSS
    // CHECKSTYLE_RULES.ON: CyclomaticComplexity
    // Set masked sql statement if request is not from OldApiUtils.handleOldApiRequest
    ApiServerUtils.setMaskedSqlIfNeeded(request);

    QueryLogger.info("Received: " + request.toStringWithoutQuery(), request.getMaskedKsql());
    throwIfNotConfigured();

    activenessRegistrar.updateLastRequestTime();

    try {
      CommandStoreUtil.httpWaitForCommandSequenceNumber(
          commandRunner.getCommandQueue(),
          request,
          distributedCmdResponseTimeout);

      final Map<String, Object> configProperties = request.getConfigOverrides();
      denyListPropertyValidator.validateAll(configProperties);

      final KsqlRequestConfig requestConfig =
          new KsqlRequestConfig(request.getRequestProperties());
      final List<ParsedStatement> statements = ksqlEngine.parse(request.getUnmaskedKsql());

      validator.validate(
          SandboxedServiceContext.create(securityContext.getServiceContext()),
          statements,
          new SessionProperties(
              configProperties,
              localHost,
              localUrl,
              requestConfig.getBoolean(KsqlRequestConfig.KSQL_REQUEST_INTERNAL_REQUEST),
              request.getSessionVariables()
          ),
          request.getUnmaskedKsql()
      );

      // log validated statements for query anonymization
      statements.forEach(s -> {
        if (s.getUnMaskedStatementText().toLowerCase().contains("terminate")
            || s.getUnMaskedStatementText().toLowerCase().contains("drop")) {
          QueryLogger.info("Query terminated", s.getMaskedStatementText());
        } else {
          QueryLogger.info("Query created", s.getMaskedStatementText());
        }
      });

      final KsqlEntityList entities = handler.execute(
          securityContext,
          statements,
          new SessionProperties(
              configProperties,
              localHost,
              localUrl,
              requestConfig.getBoolean(KsqlRequestConfig.KSQL_REQUEST_INTERNAL_REQUEST),
              request.getSessionVariables()
          )
      );

      QueryLogger.info(
          "Processed successfully: " + request.toStringWithoutQuery(),
          request.getMaskedKsql()
      );
      addCommandRunnerWarning(
          entities,
          commandRunnerWarning);
      return EndpointResponse.ok(entities);
    } catch (final KsqlRestException e) {
      QueryLogger.info(
          "Processed unsuccessfully: " + request.toStringWithoutQuery(),
          request.getMaskedKsql(),
          e
      );
      throw e;
    } catch (final KsqlStatementException e) {
      QueryLogger.info(
          "Processed unsuccessfully: " + request.toStringWithoutQuery(),
          request.getMaskedKsql(),
          e
      );
      final EndpointResponse response;
      if (e.getProblem() == KsqlStatementException.Problem.STATEMENT) {
        response = Errors.badStatement(e.getRawUnloggedDetails(), e.getSqlStatement());
      } else if (e.getProblem() == KsqlStatementException.Problem.OTHER) {
        response = Errors.serverErrorForStatement(e, e.getSqlStatement());
      } else {
        response = Errors.badRequest(e);
      }
      return errorHandler.generateResponse(e, response);
    } catch (final KsqlException e) {
      QueryLogger.info(
          "Processed unsuccessfully: " + request.toStringWithoutQuery(),
          request.getMaskedKsql(),
          e
      );
      return errorHandler.generateResponse(e, Errors.badRequest(e));
    } catch (final Exception e) {
      QueryLogger.info(
          "Processed unsuccessfully: " + request.toStringWithoutQuery(),
          request.getMaskedKsql(),
          e
      );
      return errorHandler.generateResponse(
          e,
          Errors.serverErrorForStatement(e, request.getMaskedKsql())
      );
    }
  }
@Test
  public void queryLoggerShouldReceiveTerminateStatementsWhenHandleKsqlStatementWithTerminate() {
    // Given:
    final PersistentQueryMetadata queryMetadata = createQuery(
        "CREATE STREAM test_explain AS SELECT * FROM test_stream;",
        emptyMap());
    final String terminateSql = "TERMINATE " + queryMetadata.getQueryId() + ";";

    // When:
    try (MockedStatic<QueryLogger> logger = Mockito.mockStatic(QueryLogger.class)) {
      ksqlResource.handleKsqlStatements(securityContext, new KsqlRequest(
          terminateSql,
          ImmutableMap.of(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"),
          emptyMap(),
          0L));

      // Then:
      logger.verify(() -> QueryLogger.info("Query terminated",
          terminateSql), times(1));
    }
  }