Refactor benchmark dispatch from per-tenant to orchestrator-level using Flux concurrency

[xinlian12]

Problem

Currently each tenant/benchmark independently controls its own concurrency and max operation count (via per-tenant Flux/ExecutorService). As more tenants are added, maintaining concurrency config across all of them becomes painful and error-prone.

Solution

Centralize dispatch at the orchestrator level: a single Flux.generate().flatMap(benchmark.next(), globalConcurrency) drives all operations, randomly selecting which tenant executes each slot.

Architecture

BEFORE:  Orchestrator -> ExecutorService (1 thread/tenant) -> each benchmark.run() with per-tenant concurrency
AFTER:   Orchestrator -> Flux.generate(totalOps).flatMap(randomBenchmark.next(), globalConcurrency)

Changes

Core Design

• Benchmark.java - Interface now has Mono<Void> next() (execute one operation) + shutdown(). Removed run() and supportsOrchestratorDispatch().
• BenchmarkOrchestrator.java - runWorkload() uses a single Flux with ThreadLocalRandom tenant selection and doFinally for operation counting. Removed ExecutorService.
• BenchmarkConfig.java - concurrency, numberOfOperations, and maxRunningTimeDuration are now required top-level fields in the workload config JSON. Validates concurrency > 0.

Benchmark Adaptations

• AsyncBenchmark.java - next() delegates to performWorkload() via AtomicLong counter. Removed per-benchmark Scheduler.
• SyncBenchmark.java - next() wraps blocking performWorkload() in Mono.fromCallable().subscribeOn(Schedulers.boundedElastic()). Removed ExecutorService, Semaphore, ResultHandler.
• AsyncEncryptionBenchmark.java / AsyncCtlWorkload.java - Same pattern, removed scheduler.
• LICtlWorkload.java - next() wraps full lifecycle in Mono.fromRunnable().

Config Format

{
  "concurrency": 100,
  "numberOfOperations": 500000,
  "tenantDefaults": { ... },
  "tenants": [ ... ]
}

Testing

• mvn compile test-compile -pl . -am -DskipTests passes clean
• Updated WorkflowTest and ReadMyWritesConsistencyTest for new constructor signatures

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Refactors benchmark execution so the orchestrator centrally controls concurrency and operation dispatch using Reactor Flux/Mono, while retaining a legacy per-benchmark run() path for benchmarks that can’t participate.

Changes:

• Introduces Benchmark.next() + supportsOrchestratorDispatch() to enable orchestrator-driven, per-operation dispatch.
• Reworks BenchmarkOrchestrator.runWorkload() to partition benchmarks and drive orchestrator-capable ones via Flux.generate(...).flatMap(..., globalConcurrency).
• Adds orchestrator-level config (concurrency, numberOfOperations, maxRunningTimeDuration) with fallbacks, and updates the sample workload JSON.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
sdk/cosmos/azure-cosmos-benchmark/workload-config-sample.json	Adds example orchestrator section for centralized concurrency/ops.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/Benchmark.java	Adds next() + capability flag for orchestrator dispatch.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncBenchmark.java	Implements next() for async benchmarks using an internal counter.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/SyncBenchmark.java	Implements next() for sync benchmarks via Mono.fromCallable(...).boundedElastic().
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/encryption/AsyncEncryptionBenchmark.java	Implements next() for encryption benchmark similarly to AsyncBenchmark.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/ctl/AsyncCtlWorkload.java	Implements next() to allow orchestrator dispatch for CTL async workload.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/BenchmarkConfig.java	Adds orchestrator-level config parsing + fallback getters.
sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/BenchmarkOrchestrator.java	Centralizes concurrency and operation dispatch via Flux; keeps legacy path for unsupported benchmarks.

[xinlian12]

@sdkReviewAgent

[xinlian12]

@sdkReviewAgent-2

[xinlian12]

✅ Review complete (43:40)

Posted 5 inline comment(s).

_{Steps: ✓ context, correctness, cross-sdk, design, history, past-prs, synthesis, test-coverage}