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 shouldDistributeProperties() {
    // Given
    givenRequestHandler(ImmutableMap.of());
    when(sessionProperties.getMutableScopedProperties()).thenReturn(ImmutableMap.of("x", "y"));
    // When
    final List<ParsedStatement> statements =
        KSQL_PARSER.parse(SOME_STREAM_SQL);
    final KsqlEntityList entities = handler.execute(
        securityContext,
        statements,
        sessionProperties
    );

    // Then
    assertThat(entities, contains(entity));
    verify(distributor, times(2))
        .execute(
            argThat(is(configured(
                preparedStatement(instanceOf(CreateStream.class)),
                    ImmutableMap.of("x", "y"),
                    ksqlConfig))),
            eq(ksqlEngine),
            eq(securityContext)
        );
  }