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 shouldAbortOnError_ProducerFencedException() {
    // When:
    doThrow(new ProducerFencedException("Error!")).when(transactionalProducer).commitTransaction();
    final KsqlStatementException e = assertThrows(
        KsqlStatementException.class,
        () -> distributor.execute(CONFIGURED_STATEMENT, executionContext, securityContext)
    );

    assertThat(e.getMessage(), containsString("Could not write the statement into the command topic."));
    assertThat(e.getUnloggedMessage(), containsString("Could not write the statement "
        + "'statement' into the command topic."));
    assertThat(e.getSqlStatement(), containsString("statement"));

    // Then:
    verify(queue).abortCommand(IDGEN.getCommandId(CONFIGURED_STATEMENT.getStatement()));
  }