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 shouldEnqueueSuccessfulCommandTransactionally() {
    // When:
    distributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext);

    // Then:
    final InOrder inOrder = Mockito.inOrder(transactionalProducer, queue, validatedCommandFactory);
    inOrder.verify(transactionalProducer).initTransactions();
    inOrder.verify(transactionalProducer).beginTransaction();
    inOrder.verify(queue).waitForCommandConsumer();
    inOrder.verify(validatedCommandFactory).create(
        CONFIGURED_STATEMENT,
        sandboxContext
    );
    inOrder.verify(queue).enqueueCommand(
        IDGEN.getCommandId(CONFIGURED_STATEMENT.getStatement()),
        command,
        transactionalProducer
    );
    inOrder.verify(transactionalProducer).commitTransaction();
    inOrder.verify(transactionalProducer).close();
  }