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 shouldWaitForDistributedStatements() {
    // Given
    final KsqlEntity entity1 = mock(KsqlEntity.class);
    final KsqlEntity entity2 = mock(KsqlEntity.class);
    final KsqlEntity entity3 = mock(KsqlEntity.class);

    final StatementExecutor<CreateStream> customExecutor = givenReturningExecutor(
        CreateStream.class, entity1, entity2, entity3);
    givenRequestHandler(
        ImmutableMap.of(CreateStream.class, customExecutor)
    );

    final List<ParsedStatement> statements =
        KSQL_PARSER.parse(
            "CREATE STREAM x WITH (value_format='json', kafka_topic='x');"
                + "CREATE STREAM y WITH (value_format='json', kafka_topic='y');"
                + "CREATE STREAM z WITH (value_format='json', kafka_topic='z');"
        );

    // When
    handler.execute(securityContext, statements, sessionProperties);

    // Then
    verify(sync).waitFor(argThat(hasItems(entity1, entity2)), any());
    // since the entities passed into sync#waitFor are always the same object, mockito
    // cannot verify the original two arguments
    verify(sync, times(3)).waitFor(any(), any());
  }