public int validate(
      final ServiceContext serviceContext,
      final List<ParsedStatement> statements,
      final SessionProperties sessionProperties,
      final String sql
  ) {
    requireSandbox(serviceContext);

    final KsqlExecutionContext ctx = requireSandbox(snapshotSupplier.apply(serviceContext));
    final Injector injector = injectorFactory.apply(ctx, serviceContext);
    final KsqlConfig ksqlConfig = ctx.getKsqlConfig();

    int numPersistentQueries = 0;
    for (final ParsedStatement parsed : statements) {
      final PreparedStatement<?> prepared = ctx.prepare(
          parsed,
          (isVariableSubstitutionEnabled(sessionProperties, ksqlConfig)
              ? sessionProperties.getSessionVariables()
              : Collections.emptyMap())
      );
      final ConfiguredStatement<?> configured = ConfiguredStatement.of(prepared,
          SessionConfig.of(ksqlConfig, sessionProperties.getMutableScopedProperties())
      );

      final int currNumPersistentQueries = validate(
          serviceContext,
          configured,
          sessionProperties,
          ctx,
          injector
      );
      numPersistentQueries += currNumPersistentQueries;

      if (currNumPersistentQueries > 0
          && QueryCapacityUtil.exceedsPersistentQueryCapacity(ctx, ksqlConfig)) {
        QueryCapacityUtil.throwTooManyActivePersistentQueriesException(ctx, ksqlConfig, sql);
      }
    }

    return numPersistentQueries;
  }
@SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT")
  @Test
  public void shouldCallPrepareStatementWithSessionVariables() {
    // Given
    givenRequestValidator(ImmutableMap.of(CreateStream.class, statementValidator));
    final Map<String, String> sessionVariables = ImmutableMap.of("a", "1");
    when(sessionProperties.getSessionVariables()).thenReturn(sessionVariables);
    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_VARIABLE_SUBSTITUTION_ENABLE)).thenReturn(true);

    // When
    final List<ParsedStatement> statements = givenParsed(SOME_STREAM_SQL);
    validator.validate(serviceContext, statements, sessionProperties, "sql");

    // Then
    verify(sandboxEngine).prepare(statements.get(0), sessionVariables);
    verify(sessionProperties).getSessionVariables();
  }