public KsqlEntityList execute(
      final KsqlSecurityContext securityContext,
      final List<ParsedStatement> statements,
      final SessionProperties sessionProperties
  ) {
    final KsqlEntityList entities = new KsqlEntityList();
    for (final ParsedStatement parsed : statements) {
      final PreparedStatement<?> prepared = ksqlEngine.prepare(
          parsed,
          (isVariableSubstitutionEnabled(sessionProperties)
              ? sessionProperties.getSessionVariables()
              : Collections.emptyMap())
      );

      executeStatement(
          securityContext,
          prepared,
          sessionProperties,
          entities
      ).ifPresent(entities::add);
    }
    return entities;
  }
@Test
  public void shouldThrowOnCreateStreamIfFeatureFlagIsDisabled() {
    // Given
    final StatementExecutor<CreateStream> customExecutor =
        givenReturningExecutor(CreateStream.class, mock(KsqlEntity.class));
    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_HEADERS_COLUMNS_ENABLED)).thenReturn(false);
    givenRequestHandler(ImmutableMap.of(CreateStream.class, customExecutor));

    // When
    final List<ParsedStatement> statements =
        KSQL_PARSER.parse("CREATE STREAM x (c1 ARRAY<STRUCT<`KEY` STRING, `VALUE` BYTES>> HEADERS) "
            + "WITH (value_format='json', kafka_topic='x');");
    final Exception e = assertThrows(
        KsqlException.class,
        () -> handler.execute(securityContext, statements, sessionProperties));

    // Then
    assertThat(e.getMessage(), containsString(
        "Cannot create Stream because schema with headers columns is disabled."));
  }