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

    // Then:
    verify(schemaInjector, times(1)).inject(eq(CONFIGURED_STATEMENT));
  }