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 shouldThrowExceptionWhenInsertIntoReadOnlyTopic() {
    // Given
    final PreparedStatement<Statement> preparedStatement =
        PreparedStatement.of("", new InsertInto(SourceName.of("s1"), mock(Query.class)));
    final ConfiguredStatement<Statement> configured =
        ConfiguredStatement.of(preparedStatement, SessionConfig.of(KSQL_CONFIG, ImmutableMap.of())
        );
    final DataSource dataSource = mock(DataSource.class);
    doReturn(dataSource).when(metaStore).getSource(SourceName.of("s1"));
    when(dataSource.getKafkaTopicName()).thenReturn("_confluent-ksql-default__command-topic");

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> distributor.execute(configured, executionContext, mock(KsqlSecurityContext.class))
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Cannot insert into read-only topic: "
            + "_confluent-ksql-default__command-topic"));
  }