public StatementExecutorResponse execute(
      final ConfiguredStatement<? extends Statement> statement,
      final KsqlExecutionContext executionContext,
      final KsqlSecurityContext securityContext
  ) {
    final String commandRunnerWarningString = commandRunnerWarning.get();
    if (!commandRunnerWarningString.equals("")) {
      throw new KsqlServerException("Failed to handle Ksql Statement."
          + System.lineSeparator()
          + commandRunnerWarningString);
    }
    final InjectorWithSideEffects injector = InjectorWithSideEffects.wrap(
            injectorFactory.apply(executionContext, securityContext.getServiceContext()));
    final ConfiguredStatementWithSideEffects<?> injectedWithSideEffects =
            injector.injectWithSideEffects(statement);
    try {
      return executeInjected(
            injectedWithSideEffects.getStatement(),
            statement,
            executionContext,
            securityContext);
    } catch (Exception e) {
      injector.revertSideEffects(injectedWithSideEffects);
      throw e;
    }
  }
@Test
  public void shouldThrowIfRateLimitHit() {
    // Given:
    final DistributingExecutor rateLimitedDistributor = new DistributingExecutor(
      new KsqlConfig(ImmutableMap.of(KsqlRestConfig.KSQL_COMMAND_TOPIC_RATE_LIMIT_CONFIG, 0.25)),
      queue,
      DURATION_10_MS,
      (ec, sc) -> InjectorChain.of(schemaInjector, topicInjector),
      Optional.of(authorizationValidator),
      validatedCommandFactory,
      errorHandler,
      commandRunnerWarning
    );
    
    // When:
    rateLimitedDistributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);


    // Then:
    boolean exceptionFound = false;
    try {
      rateLimitedDistributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);
      rateLimitedDistributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);
      rateLimitedDistributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);
      rateLimitedDistributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);
    } catch (Exception e) {
      assertTrue(e instanceof KsqlRestException);
      final KsqlRestException restException = (KsqlRestException) e; 
      assertEquals(restException.getResponse().getStatus(), 429);
      final KsqlErrorMessage errorMessage = (KsqlErrorMessage) restException.getResponse().getEntity();
      assertTrue(errorMessage.getMessage().contains("DDL/DML rate is crossing the configured rate limit of statements/second"));
      exceptionFound = true;
    }
    assertTrue(exceptionFound);
  }