From 1b89dc93d4239eb0074f5ce7302be77011bfbca9 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 31 Aug 2026 12:52:21 -0700
Subject: [PATCH 1/2] bench(soak): add mysql_cdc to the nightly soak rotation
 (CON-179 R6)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Second connector in the soak rotation, following SOAK.md's recipe:

- scenarios/mysql/orders-soak.yaml: the same sustained-moderate profile as
  the postgres soak (10K writes/sec x 1.2 KB ~= 12 MB/s, ~10-15% of
  mysql_cdc's ~100-115 MB/s measured read ceiling) for 90 minutes at one
  vCPU point, plus the 30m orders-soak-pr.yaml binary-arm variant for
  /soak mysql/orders-soak-pr. checkpoint_limit is raised above
  batching.count (the 1024 default admits an oversized batch only when
  nothing is pending, serializing the pipeline to one batch in flight),
  and the reset sets RDS 'binlog retention hours' every run —
  backup_retention_period only ENABLES binlog; RDS purges backed-up
  binlogs within minutes without the rds_set_configuration call, and a
  mid-soak reconnect would die with ERROR 1236.
- terraform stacks/mysql + modules/rds-mysql: recovered from the
  benchmarking branch and brought up to this tree's conventions
  (bench-session-id tag for the reaper, cloud-nuke-excluded exemption,
  required_version 1.10 for lockfile locking, nullable storage_throughput
  with the same >=400 GB gp3 tiering as postgres). binlog_format=ROW /
  binlog_row_image=FULL are what mysql_cdc requires.
- runner: the mysql_cdc engineSpec entry and the discrete-flags reset
  form return from git history, with the current rpk-based topic cleanup
  kept. The reset renders `mariadb` (guaranteed by mariadb1011 on the
  runner host) rather than betting on the deprecated `mysql` compat
  symlink, refuses at render time to emit a command with any empty
  connection part (bash collapses -p"" to a bare -p an hour into a paid
  session), and rejects multiline reset SQL, which %q quoting would
  silently alter.
- seeders/cdc-rows-mysql: a port of the current postgres seeder — the old
  benchmarking-branch mysql seeder predates the worker/tick remainder
  fairness fixes and the one-placeholder-per-row change, so this ports
  forward instead of copying back. InterpolateParams is enabled because,
  unlike pgx, go-sql-driver otherwise wraps every parameterized Exec in a
  hidden server-side Prepare+Execute+Close.
- soak_nightly.yml: the rotation becomes a matrix resolved by a plan job
  (the dispatch input is allowlisted BEFORE strategy evaluation, so a
  crafted value can't fan one dispatch into N paid soak jobs), serialized
  with max-parallel: 1 — one bench at a time, ever; manual dispatch still
  runs exactly the scenario named in the input.
- terraform/persistent soak_scenarios gains the mysql entry (dashboard +
  stall/rss-slope/backlog alarms per entry).

Password auth deliberately: the IAM credential-rotation lever this soak
is destined for (RDS IAM tokens ride normal-protocol MySQL connections,
unlike postgres replication) lands as its own increment — it needs the
AWSAuthenticationPlugin DB user in reset, an rds-db:connect policy in the
stack, and a live validation pass. SOAK.md/README updated accordingly.

Operator steps BEFORE this merges (the cron arms itself on the default
branch): `task aws:persistent` re-apply (new dashboard + alarms), then a
manual nightly dispatch with scenario=mysql/orders-soak; the baseline
comparator stays advisory until three soak-index entries exist.

Known follow-ups (shared with the postgres seeder, deferred to keep the
two in lockstep): extract the duplicated worker/tick/payload math into a
shared package, and cancel sibling workload workers on first error
instead of degrading to 15/16 rate for the rest of the window.

Verified: go test -race -shuffle=on, golangci-lint, gofumpt clean on the
benchmarking module; terraform validate + fmt clean on the new stack;
both scenarios pass `task aws:validate`.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .github/workflows/soak_nightly.yml            |  61 ++-
 benchmarking/aws/README.md                    |  21 +-
 benchmarking/aws/SOAK.md                      |  28 +-
 benchmarking/aws/go.mod                       |   2 +
 benchmarking/aws/go.sum                       |   4 +
 benchmarking/aws/runner/scenario.go           |  15 +-
 benchmarking/aws/runner/scenario_test.go      |  67 ++-
 benchmarking/aws/runner/scripts.go            |  52 ++-
 benchmarking/aws/runner/scripts_test.go       |  92 ++++
 .../aws/scenarios/mysql/orders-soak-pr.yaml   |  80 ++++
 .../aws/scenarios/mysql/orders-soak.yaml      |  96 +++++
 .../aws/seeders/cdc-rows-mysql/main.go        |  52 +++
 .../aws/seeders/cdc-rows-mysql/sql.go         | 393 ++++++++++++++++++
 .../aws/seeders/cdc-rows-mysql/sql_test.go    | 153 +++++++
 .../aws/terraform/modules/rds-mysql/main.tf   |  85 ++++
 .../terraform/modules/rds-mysql/outputs.tf    |  16 +
 .../terraform/modules/rds-mysql/variables.tf  |  51 +++
 .../aws/terraform/persistent/variables.tf     |   4 +
 .../aws/terraform/stacks/mysql/main.tf        |  54 +++
 .../aws/terraform/stacks/mysql/outputs.tf     |  20 +
 .../aws/terraform/stacks/mysql/variables.tf   |  34 ++
 21 files changed, 1342 insertions(+), 38 deletions(-)
 create mode 100644 benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml
 create mode 100644 benchmarking/aws/scenarios/mysql/orders-soak.yaml
 create mode 100644 benchmarking/aws/seeders/cdc-rows-mysql/main.go
 create mode 100644 benchmarking/aws/seeders/cdc-rows-mysql/sql.go
 create mode 100644 benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go
 create mode 100644 benchmarking/aws/terraform/modules/rds-mysql/main.tf
 create mode 100644 benchmarking/aws/terraform/modules/rds-mysql/outputs.tf
 create mode 100644 benchmarking/aws/terraform/modules/rds-mysql/variables.tf
 create mode 100644 benchmarking/aws/terraform/stacks/mysql/main.tf
 create mode 100644 benchmarking/aws/terraform/stacks/mysql/outputs.tf
 create mode 100644 benchmarking/aws/terraform/stacks/mysql/variables.tf

diff --git a/.github/workflows/soak_nightly.yml b/.github/workflows/soak_nightly.yml
index 7862cdd0e6..1d905860cb 100644
--- a/.github/workflows/soak_nightly.yml
+++ b/.github/workflows/soak_nightly.yml
@@ -49,22 +49,66 @@ permissions:
   contents: read
 
 jobs:
+  # Resolve the scenario matrix BEFORE strategy evaluation ever touches the
+  # dispatch input. Splicing the raw input into fromJSON(format(...)) would
+  # let a crafted value (`a","b`) fan one dispatch into N paid soak jobs,
+  # and a stray quote would kill the run with an opaque strategy error
+  # instead of the allowlist's friendly one — so the input is allowlisted
+  # here (quotes and backslashes can't pass the regex) and the soak job
+  # consumes only this job's output.
+  plan:
+    runs-on: ubuntu-latest
+    outputs:
+      scenarios: ${{ steps.set.outputs.scenarios }}
+    steps:
+      - name: Resolve scenario matrix
+        id: set
+        env:
+          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+        run: |
+          if [ "${{ github.event_name }}" = "schedule" ]; then
+            # THE ROTATION lives here: the schedule soaks every listed
+            # scenario; adding a connector to the rotation = one more entry
+            # (see SOAK.md "Adding a connector to the rotation").
+            echo 'scenarios=["postgres/orders-soak", "mysql/orders-soak"]' >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+          if ! echo "${SCENARIO}" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'; then
+            echo "::error::Invalid scenario '${SCENARIO}' — expected a path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
+            exit 1
+          fi
+          echo "scenarios=[\"${SCENARIO}\"]" >> "$GITHUB_OUTPUT"
+
   soak:
+    needs: plan
+    strategy:
+      # max-parallel: 1 because one bench at a time, ever — all sessions
+      # share (and destroy) the same shared Terraform stack and fixed-name
+      # session resources. The workflow-level concurrency group serializes
+      # across runs; this serializes within one.
+      #
+      # fail-fast: false so one connector's red night still soaks the rest.
+      max-parallel: 1
+      fail-fast: false
+      matrix:
+        scenario: ${{ fromJSON(needs.plan.outputs.scenarios) }}
     runs-on: ubuntu-latest
     # Provisioning (~15m) + 95m window + teardown (~15m) + slack. Must stay
-    # under the 4h credential session below.
+    # under the 4h credential session below. Per matrix job, so the full
+    # scheduled rotation may take rotation-size × this.
     timeout-minutes: 220
     env:
       REDPANDA_LICENSE_SECRET: redpanda-connect-bench/license
     steps:
       # Before anything credentialed: the dispatch input is available to
       # the whole write-access population (same as soak_pr.yml's /soak
-      # comment), and Task splices {{.scenario}} raw into a shell command —
-      # so it gets the same allowlist soak_pr.yml applies, and an invalid
-      # value kills the job before the provisioner role is even assumed.
+      # comment), and Task splices {{.scenario}} raw into a shell command.
+      # The plan job already allowlisted it, so this is defense in depth —
+      # it re-checks the value each job actually received, and still kills
+      # the job before the provisioner role is even assumed.
       - name: Validate scenario input
         env:
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
           if ! echo "${SCENARIO}" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'; then
             echo "::error::Invalid scenario '${SCENARIO}' — expected a path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
@@ -91,7 +135,7 @@ jobs:
         id: gate
         shell: bash
         env:
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
           if [ "${{ github.event_name }}" != "schedule" ]; then
             echo "run=true" >> "$GITHUB_OUTPUT"
@@ -156,7 +200,7 @@ jobs:
         env:
           # Environment passing, not ${{ }} substitution into the script —
           # same reasoning as the gate step above and soak_pr.yml.
-          SCENARIO: ${{ github.event.inputs.scenario || 'postgres/orders-soak' }}
+          SCENARIO: ${{ matrix.scenario }}
         run: |
           task aws:bench scenario="$SCENARIO" 2>&1 | tee "$RUNNER_TEMP/soak-run.log"
 
@@ -182,6 +226,7 @@ jobs:
         if: always() && steps.gate.outputs.run == 'true'
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
         with:
-          name: soak-run-log
+          # job-index, not the scenario path: artifact names reject `/`.
+          name: soak-run-log-${{ strategy.job-index }}
           path: ${{ runner.temp }}/soak-run.log
           retention-days: 30
diff --git a/benchmarking/aws/README.md b/benchmarking/aws/README.md
index 800c9725c1..d71738d13e 100644
--- a/benchmarking/aws/README.md
+++ b/benchmarking/aws/README.md
@@ -3,11 +3,11 @@
 Production-shaped benchmarks and soak tests for Redpanda Connect connectors,
 run on real AWS infrastructure in a dedicated, disposable account.
 
-This tree contains the framework core plus the **postgres_cdc** stack — the
-subset needed by the soak pipeline (CON-179 R6). Further connector stacks
-(mysql, sqlserver, oracle, mongodb, dynamodb, iceberg) exist on the original
-development fork and land here with their own PRs, each bringing its
-scenarios and tests.
+This tree contains the framework core plus the **postgres_cdc** and
+**mysql_cdc** stacks — the subset needed by the soak pipeline (CON-179 R6).
+Further connector stacks (sqlserver, oracle, mongodb, dynamodb, iceberg)
+exist on the original development fork and land here with their own PRs,
+each bringing its scenarios and tests.
 
 ## What a run does
 
@@ -90,10 +90,10 @@ and re-enabled after.
 | Path | Role |
 |---|---|
 | `runner/` | Go orchestrator: provision → stage → seed → sweep/soak → results → teardown |
-| `scenarios/postgres/` | bench + soak + PR-comparison scenarios |
-| `seeders/cdc-rows-postgres/` | write-workload generator |
+| `scenarios/<engine>/` | bench + soak + PR-comparison scenarios (postgres, mysql) |
+| `seeders/cdc-rows-<engine>/` | write-workload generators |
 | `terraform/shared/` | per-session VPC, hosts, brokers, results bucket |
-| `terraform/stacks/postgres/` | per-session RDS Postgres |
+| `terraform/stacks/<engine>/` | per-session RDS Postgres / RDS MySQL |
 | `terraform/persistent/` | applied once: dashboards, alarms, OIDC, reaper, archive |
 | `cleanup-lambda/` | the orphan reaper (own Go module) |
 | `SOAK.md` | soak operations runbook |
@@ -101,8 +101,9 @@ and re-enabled after.
 ## Known limitations
 
 - postgres_cdc IAM auth cannot work against vanilla RDS (replication
-  connections reject IAM tokens); the credential-rotation soak window is
-  covered by mysql_cdc when its stack lands.
+  connections reject IAM tokens); the credential-rotation soak window
+  belongs to the mysql_cdc soak, which runs password auth today — the IAM
+  increment is still open (see SOAK.md).
 - One-lane serialization: soaks and benches queue on the shared stack.
   Session-scoped isolation is the tracked scaling path.
 - The weekly 24h soak needs a reaper exemption tag + a non-GitHub conductor
diff --git a/benchmarking/aws/SOAK.md b/benchmarking/aws/SOAK.md
index 1eaeea09a4..47f1b99d88 100644
--- a/benchmarking/aws/SOAK.md
+++ b/benchmarking/aws/SOAK.md
@@ -15,7 +15,7 @@ files named below.
 | Runner soak mode | `runner/` (main.go, matrix.go, cloudwatch.go) | scaled cadences, 10-min S3 checkpoints, per-minute CloudWatch emission, backlog series |
 | Dashboards + alarms | `terraform/persistent/` (`main.tf` `soak_scenarios` var, `alarms.tf`) | one dashboard + three alarms (stall / rss-slope / backlog) per scenario → SNS `redpanda-connect-bench-soak-alerts` |
 | Archive + baseline | `redpanda-connect-bench-soak-archive` bucket | result.json + raw artifacts per run; `soak-index/` feeds the rolling-baseline comparator (advisory < 3 runs, then fails the job on throughput < 85% / RSS > 130% of baseline) |
-| Nightly workflow | `.github/workflows/soak_nightly.yml` | 08:10 UTC cron (arms only from the default branch) + manual dispatch; OIDC creds (4h), license from Secrets Manager, teardown verified against AWS |
+| Nightly workflow | `.github/workflows/soak_nightly.yml` | 08:10 UTC cron over the rotation matrix (postgres, mysql; serialized, arms only from the default branch) + manual dispatch; OIDC creds (4h), license from Secrets Manager, teardown verified against AWS |
 | PR comparison | `.github/workflows/soak_pr.yml` | `/soak` comment (write-access gated) → base-vs-PR binaries, same infra, sticky comparison comment |
 
 ## Adding a connector to the rotation
@@ -32,14 +32,23 @@ files named below.
    in `terraform/persistent/variables.tf` (key → connector + scenario
    name), then `task aws:persistent`. Alarms and the dashboard are
    generated per entry; Slack delivery is on by default via `slack.tf`'s
-   committed IDs — no extra vars needed.
-4. **First runs**: dispatch the nightly workflow manually with the
+   committed IDs — no extra vars needed. **Apply this BEFORE the rotation
+   entry merges** — the cron arms itself the moment the workflow lands on
+   the default branch, and until the persistent apply runs, nothing
+   watches the new connector's metrics.
+4. **Add it to the nightly rotation**: append the scenario path to the
+   schedule list in `soak_nightly.yml`'s plan job. The matrix is
+   serialized (`max-parallel: 1` — one bench at a time) and
+   `fail-fast: false`, so each rotation entry soaks even when another is
+   red; a manual dispatch still runs only the scenario named in the input.
+5. **First runs**: dispatch the nightly workflow manually with the
    scenario input. The baseline comparator stays advisory until three
    soak-index entries exist.
-5. **Optionally add a PR variant** (`*-soak-pr.yaml`): same scenario with
+6. **Optionally add a PR variant** (`*-soak-pr.yaml`): same scenario with
    30m duration and `arms: [{id: base, binary: base}, {id: pr, binary:
    pr}]`. The scenario NAME must differ from the nightly's so its metrics
-   land outside the alarm dimensions.
+   land outside the alarm dimensions. `/soak <engine>/<name>-pr` selects it
+   on a PR (the default remains postgres/orders-soak-pr).
 
 ## Operating it
 
@@ -106,7 +115,7 @@ files named below.
   backend now uses S3-native `use_lockfile` locking), disarmed the orphan
   reaper's schedule rule **every night**, and deleted the stall + backlog
   alarms. The exemption is the `cloud-nuke-excluded = true` tag, applied via
-  `default_tags` in all three stacks — the persistent stack so the reaper
+  `default_tags` in every stack — the persistent stack so the reaper
   schedule and alarms survive, the session stacks so a live bench crossing
   02:25 UTC isn't terminated mid-run. Our OWN reaper keys on `Project`, not
   this tag, so bench cleanup at the 4h TTL is unaffected. Any new resource
@@ -115,8 +124,11 @@ files named below.
 
 - postgres_cdc IAM auth cannot work against vanilla RDS (replication-
   protocol connections reject IAM tokens — verified live 2026-08-12), so
-  the credential-rotation window is covered by a future mysql_cdc soak or
-  Aurora, not the postgres soak.
+  the postgres soak cannot cover the credential-rotation window. The mysql
+  soak is its designated home (RDS IAM tokens ride normal-protocol MySQL
+  connections), but it runs password auth today — the IAM increment (the
+  AWSAuthenticationPlugin DB user in reset, the rds-db:connect policy in
+  the mysql stack, a live validation pass) is still open.
 - The nightly cron only arms once `soak_nightly.yml` is on the repo's
   default branch; until then, manual dispatch.
 - The weekly 24h soak needs two prerequisites before it can exist: a
diff --git a/benchmarking/aws/go.mod b/benchmarking/aws/go.mod
index e447aeab91..eb13c388d3 100644
--- a/benchmarking/aws/go.mod
+++ b/benchmarking/aws/go.mod
@@ -18,6 +18,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.44.6
 	github.com/aws/aws-sdk-go-v2/service/ssm v1.73.6
 	github.com/aws/smithy-go v1.27.8
+	github.com/go-sql-driver/mysql v1.10.0
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23
 	github.com/stretchr/testify v1.12.1
@@ -25,6 +26,7 @@ require (
 )
 
 require (
+	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.36 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.37 // indirect
diff --git a/benchmarking/aws/go.sum b/benchmarking/aws/go.sum
index c7eee17ed2..cccab5f944 100644
--- a/benchmarking/aws/go.sum
+++ b/benchmarking/aws/go.sum
@@ -1,3 +1,5 @@
+filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
+filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/aws/aws-sdk-go-v2 v1.43.6 h1:RrmFcqCBxkJuf7g1axVo5krB4jM/AO8r5e5oujrgdoQ=
 github.com/aws/aws-sdk-go-v2 v1.43.6/go.mod h1:tXpPM+v0D1lndmga+HqqLDIzUFJlEeR21aspVklHF00=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 h1:LAfOuhAH331fmOjTQpAaOlH+Ftn7RzSDJ2VFwjdMMy4=
@@ -48,6 +50,8 @@ github.com/aws/smithy-go v1.27.8 h1:FR0dxZfIlV7Z8eh2iHfIofdunw382XsDV3Mxt9nUvRY=
 github.com/aws/smithy-go v1.27.8/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw=
+github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
diff --git a/benchmarking/aws/runner/scenario.go b/benchmarking/aws/runner/scenario.go
index 9858df628a..4bb743d3ce 100644
--- a/benchmarking/aws/runner/scenario.go
+++ b/benchmarking/aws/runner/scenario.go
@@ -207,7 +207,7 @@ type engineSpec struct {
 }
 
 // engineSpecs is the registry mechanism a connector's stack PR extends —
-// see the type doc above. mysql_cdc, oracledb_cdc, microsoft_sql_server_cdc,
+// see the type doc above. oracledb_cdc, microsoft_sql_server_cdc,
 // mongodb_cdc, and aws_dynamodb_cdc were trimmed out of this scope-reduced
 // tree (postgres_cdc soak testing only); each returns with its own stack PR.
 var engineSpecs = map[string]engineSpec{
@@ -215,6 +215,19 @@ var engineSpecs = map[string]engineSpec{
 		DSNOutputKey: "postgres_dsn",
 		DSNEnvVar:    "POSTGRES_DSN",
 	},
+	// mysql_cdc connects via a go-sql-driver DSN, but the mariadb CLI the
+	// reset runs through takes discrete -h/-P/-u/-p flags rather than a DSN
+	// URL, so the Reset*OutputKey fields point at the mysql stack's
+	// split-out outputs.
+	"mysql_cdc": {
+		DSNOutputKey:       "mysql_dsn",
+		DSNEnvVar:          "MYSQL_DSN",
+		ResetHostOutputKey: "mysql_host",
+		ResetPortOutputKey: "mysql_port",
+		ResetUserOutputKey: "mysql_user",
+		ResetPassOutputKey: "mysql_password",
+		ResetDBOutputKey:   "mysql_db",
+	},
 }
 
 func engineSpecFor(connector string) (engineSpec, bool) {
diff --git a/benchmarking/aws/runner/scenario_test.go b/benchmarking/aws/runner/scenario_test.go
index 8fcdac93e5..384b691dce 100644
--- a/benchmarking/aws/runner/scenario_test.go
+++ b/benchmarking/aws/runner/scenario_test.go
@@ -97,14 +97,40 @@ func TestEngineSpecFor_Postgres(t *testing.T) {
 	}
 }
 
+func TestEngineSpecFor_MySQL(t *testing.T) {
+	es, ok := engineSpecFor("mysql_cdc")
+	if !ok {
+		t.Fatalf("mysql_cdc should be registered")
+	}
+	if es.DSNOutputKey != "mysql_dsn" {
+		t.Errorf("DSNOutputKey = %q, want mysql_dsn", es.DSNOutputKey)
+	}
+	if es.DSNEnvVar != "MYSQL_DSN" {
+		t.Errorf("DSNEnvVar = %q, want MYSQL_DSN", es.DSNEnvVar)
+	}
+	// The mysql CLI takes discrete flags, not a DSN URL, so the reset
+	// builder needs every split-out output key populated.
+	for name, got := range map[string]string{
+		"ResetHostOutputKey": es.ResetHostOutputKey,
+		"ResetPortOutputKey": es.ResetPortOutputKey,
+		"ResetUserOutputKey": es.ResetUserOutputKey,
+		"ResetPassOutputKey": es.ResetPassOutputKey,
+		"ResetDBOutputKey":   es.ResetDBOutputKey,
+	} {
+		if got == "" {
+			t.Errorf("%s must be set for mysql's discrete-flags reset form", name)
+		}
+	}
+}
+
 func TestEngineSpecFor_Unknown(t *testing.T) {
 	if _, ok := engineSpecFor("kafka_franz_in_disguise"); ok {
 		t.Error("unknown connector should not resolve")
 	}
-	// mysql_cdc, oracledb_cdc, microsoft_sql_server_cdc, mongodb_cdc, and
+	// oracledb_cdc, microsoft_sql_server_cdc, mongodb_cdc, and
 	// aws_dynamodb_cdc were trimmed from the registry in this scope-reduced
-	// (postgres_cdc-only) tree; each returns with its own stack PR.
-	for _, trimmed := range []string{"mysql_cdc", "oracledb_cdc", "microsoft_sql_server_cdc", "mongodb_cdc", "aws_dynamodb_cdc"} {
+	// tree; each returns with its own stack PR.
+	for _, trimmed := range []string{"oracledb_cdc", "microsoft_sql_server_cdc", "mongodb_cdc", "aws_dynamodb_cdc"} {
 		if _, ok := engineSpecFor(trimmed); ok {
 			t.Errorf("%s should not be registered in this scope-reduced tree", trimmed)
 		}
@@ -390,6 +416,41 @@ func TestLoadScenario_OrdersSoak(t *testing.T) {
 	require.NoError(t, s.Validate())
 }
 
+// TestLoadScenario_MySQLOrdersSoak is the same validity gate for the mysql
+// rotation entry: the shipped scenarios/mysql/orders-soak.yaml must load and
+// validate as a soak scenario, not merely parse.
+func TestLoadScenario_MySQLOrdersSoak(t *testing.T) {
+	s, err := LoadScenario("../scenarios/mysql/orders-soak.yaml")
+	require.NoError(t, err)
+	require.True(t, s.Soak)
+	require.Equal(t, "mysql_cdc", s.Connector)
+	require.Equal(t, "mysql", s.Stack)
+	require.Equal(t, []int{2}, s.Matrix.CPUPoints)
+	require.Empty(t, s.Matrix.Arms)
+	require.Equal(t, "c8g.xlarge", s.Infra.Runner.InstanceType)
+	require.NotNil(t, s.Workload)
+	require.Equal(t, 90*time.Minute, s.Workload.Duration)
+	require.Equal(t, 5*time.Minute, s.Workload.Warmup)
+	require.Equal(t, 10000, s.Workload.WriteRatePerSec)
+	// mysql_cdc requires a checkpoint cache resource; the scenario must ship
+	// one or the rendered config fails lint on the runner host, an hour of
+	// provisioning too late.
+	require.Contains(t, s.Pipeline, "cache_resources")
+	require.NoError(t, s.Validate())
+}
+
+// TestLoadScenario_MySQLOrdersSoakPR pins the /soak A/B variant: binary-only
+// arms on the same single-point soak profile as the nightly, under a
+// deliberately different scenario name so its metrics dodge the alarms.
+func TestLoadScenario_MySQLOrdersSoakPR(t *testing.T) {
+	s, err := LoadScenario("../scenarios/mysql/orders-soak-pr.yaml")
+	require.NoError(t, err)
+	require.True(t, s.Soak)
+	require.True(t, s.IsBinaryArmScenario())
+	require.Equal(t, 30*time.Minute, s.Workload.Duration)
+	require.NotEqual(t, "mysql-orders-soak", s.Name)
+}
+
 func TestScenarioValidate_RejectsSoakWithMultipleCPUPoints(t *testing.T) {
 	s := &Scenario{
 		Name: "soak-x", Connector: "postgres_cdc", Stack: "postgres", Soak: true,
diff --git a/benchmarking/aws/runner/scripts.go b/benchmarking/aws/runner/scripts.go
index 151bd04c7d..f4d135b275 100644
--- a/benchmarking/aws/runner/scripts.go
+++ b/benchmarking/aws/runner/scripts.go
@@ -40,7 +40,9 @@ func envVarPrefix(es engineSpec, outs map[string]string) string {
 // shell-safe — bash still expands $, ` and \ inside "...". This is intentional
 // because (a) the RDS modules generate passwords with special=false (alphanumeric
 // only), and (b) reset SQL is hand-authored. If you ever feed user input into
-// these renderers, switch to a real shell-quoter.
+// these renderers, switch to a real shell-quoter. %q also turns a real
+// newline into a literal \n, so combineReset rejects multiline reset SQL
+// outright rather than let it reach the CLI altered.
 
 // renderSeedScript renders the shell script that runs on the load-gen host to
 // pre-seed the source database. The seeder is expected to be staged at
@@ -84,13 +86,47 @@ func combineReset(connector string, steps []ResetStep, outs map[string]string) (
 	sb.WriteString("set -euo pipefail\n")
 	for _, st := range steps {
 		if st.SQL != "" {
-			// DSN form (postgres). The discrete host/port/user/pass/db-flags
-			// form (mysql's `mysql -h ... -P ... -e ...`) was dead once the
-			// registry was trimmed to postgres_cdc only — no remaining
-			// engineSpec entry sets ResetHostOutputKey — and returns with
-			// mysql_cdc's own stack PR.
-			fmt.Fprintf(&sb, `psql %q -v ON_ERROR_STOP=1 -c %q`+"\n",
-				outs[es.DSNOutputKey], st.SQL)
+			// %q renders a newline as the two characters \n, silently
+			// altering multiline SQL before the CLI ever sees it (a comment
+			// line would swallow the rest of the statement). Refuse loudly
+			// at render time rather than corrupt quietly on the paid host.
+			if strings.ContainsAny(st.SQL, "\n\r") {
+				return "", fmt.Errorf("combineReset: reset SQL must be single-line (%%q quoting turns a newline into a literal backslash-n): %q", st.SQL)
+			}
+			if es.ResetHostOutputKey != "" {
+				// Discrete-flags form (mysql): the CLI has no DSN-URL mode.
+				// The binary is `mariadb`, not `mysql` — mariadb1011 on the
+				// runner host guarantees the former, while the deprecated
+				// mysql compat symlink can vanish in a package/AMI refresh.
+				//
+				// Every connection part must be present, or the rendered
+				// command fails only at runtime on the provisioned host
+				// (bash collapses -p"" to a bare -p prompt) — an hour of
+				// paid session too late for a renamed terraform output.
+				for _, k := range []string{
+					es.ResetHostOutputKey, es.ResetPortOutputKey,
+					es.ResetUserOutputKey, es.ResetPassOutputKey,
+					es.ResetDBOutputKey,
+				} {
+					if outs[k] == "" {
+						return "", fmt.Errorf("combineReset: connector %q: terraform output %q is missing or empty", connector, k)
+					}
+				}
+				fmt.Fprintf(&sb, `mariadb -h %q -P %q -u %q -p%q %q -e %q`+"\n",
+					outs[es.ResetHostOutputKey],
+					outs[es.ResetPortOutputKey],
+					outs[es.ResetUserOutputKey],
+					outs[es.ResetPassOutputKey],
+					outs[es.ResetDBOutputKey],
+					st.SQL)
+			} else {
+				// DSN form (postgres). Same render-time presence check.
+				if outs[es.DSNOutputKey] == "" {
+					return "", fmt.Errorf("combineReset: connector %q: terraform output %q is missing or empty", connector, es.DSNOutputKey)
+				}
+				fmt.Fprintf(&sb, `psql %q -v ON_ERROR_STOP=1 -c %q`+"\n",
+					outs[es.DSNOutputKey], st.SQL)
+			}
 		}
 		if st.Bash != "" {
 			sb.WriteString(substitutePlaceholders(st.Bash, outs) + "\n")
diff --git a/benchmarking/aws/runner/scripts_test.go b/benchmarking/aws/runner/scripts_test.go
index 92bb51432b..3bd2e0fa89 100644
--- a/benchmarking/aws/runner/scripts_test.go
+++ b/benchmarking/aws/runner/scripts_test.go
@@ -34,6 +34,24 @@ func TestRenderSeedScript_Postgres(t *testing.T) {
 	}
 }
 
+func TestRenderSeedScript_MySQL(t *testing.T) {
+	s := &Scenario{
+		Connector: "mysql_cdc",
+		Dataset:   DatasetSpec{Tables: []string{"orders"}, RowSizeBytes: 1200, Seeder: "cdc-rows-mysql", InitialRows: 0},
+	}
+	outs := map[string]string{"mysql_dsn": "u:p@tcp(host:3306)/db", "results_bucket": "bucket"}
+	script, err := renderSeedScript(s, outs, "stage/cdc-rows-mysql")
+	if err != nil {
+		t.Fatalf("renderSeedScript: %v", err)
+	}
+	if !strings.Contains(script, `MYSQL_DSN="u:p@tcp(host:3306)/db"`) {
+		t.Errorf("mysql seed script must set MYSQL_DSN; got:\n%s", script)
+	}
+	if !strings.Contains(script, "/opt/bench/cdc-rows-mysql seed") {
+		t.Errorf("mysql seed script must invoke /opt/bench/cdc-rows-mysql seed; got:\n%s", script)
+	}
+}
+
 func TestRenderSeedScript_NoDSN_WithExtraEnvVars(t *testing.T) {
 	// Register a test-only NoDSN engine and clean up after.
 	engineSpecs["aws_dynamodb_cdc_test"] = engineSpec{
@@ -92,6 +110,80 @@ func TestCombineReset_Postgres_DSNForm(t *testing.T) {
 	}
 }
 
+func TestCombineReset_MySQL_DiscreteFlagsForm(t *testing.T) {
+	steps := []ResetStep{{SQL: "TRUNCATE TABLE orders"}}
+	outs := map[string]string{
+		"mysql_dsn":      "u:p@tcp(host:3306)/db",
+		"mysql_host":     "host",
+		"mysql_port":     "3306",
+		"mysql_user":     "bench",
+		"mysql_password": "pw",
+		"mysql_db":       "benchdb",
+	}
+	got, err := combineReset("mysql_cdc", steps, outs)
+	if err != nil {
+		t.Fatalf("combineReset: %v", err)
+	}
+	// The mariadb CLI has no DSN-URL mode: every connection part must be a
+	// discrete flag, and psql must not appear. (`mariadb`, not the
+	// deprecated `mysql` compat symlink — see combineReset.)
+	if !strings.Contains(got, `mariadb -h "host" -P "3306" -u "bench" -p"pw" "benchdb" -e "TRUNCATE TABLE orders"`) {
+		t.Errorf("mysql reset must use discrete -h/-P/-u/-p flags; got:\n%s", got)
+	}
+	if strings.Contains(got, "psql") {
+		t.Errorf("mysql reset must not fall back to psql; got:\n%s", got)
+	}
+}
+
+func TestCombineReset_MySQL_RejectsMissingOutput(t *testing.T) {
+	outs := map[string]string{
+		"mysql_host": "host",
+		"mysql_port": "3306",
+		"mysql_user": "bench",
+		// mysql_password deliberately absent: bash would collapse -p"" to a
+		// bare -p prompt an hour into a paid session.
+		"mysql_db": "benchdb",
+	}
+	_, err := combineReset("mysql_cdc", []ResetStep{{SQL: "SELECT 1"}}, outs)
+	if err == nil || !strings.Contains(err.Error(), "mysql_password") {
+		t.Fatalf("expected a missing-output error naming mysql_password, got %v", err)
+	}
+}
+
+func TestCombineReset_RejectsMultilineSQL(t *testing.T) {
+	outs := map[string]string{"postgres_dsn": "postgres://u:p@host:5432/db"}
+	_, err := combineReset("postgres_cdc", []ResetStep{{SQL: "SELECT 1;\nSELECT 2"}}, outs)
+	if err == nil || !strings.Contains(err.Error(), "single-line") {
+		t.Fatalf("expected a multiline-SQL rejection, got %v", err)
+	}
+}
+
+func TestCombineReset_Postgres_RejectsMissingDSN(t *testing.T) {
+	_, err := combineReset("postgres_cdc", []ResetStep{{SQL: "SELECT 1"}}, map[string]string{})
+	if err == nil || !strings.Contains(err.Error(), "postgres_dsn") {
+		t.Fatalf("expected a missing-output error naming postgres_dsn, got %v", err)
+	}
+}
+
+func TestCombineReset_AppendsTopicCleanup_MySQL(t *testing.T) {
+	outs := map[string]string{
+		"mysql_host":                "host",
+		"mysql_port":                "3306",
+		"mysql_user":                "bench",
+		"mysql_password":            "pw",
+		"mysql_db":                  "benchdb",
+		"bench_session_id":          "sess-abc",
+		"redpanda_broker_endpoints": "10.42.10.10:9092",
+	}
+	got, err := combineReset("mysql_cdc", []ResetStep{{SQL: "TRUNCATE TABLE orders"}}, outs)
+	if err != nil {
+		t.Fatalf("combineReset: %v", err)
+	}
+	if !strings.Contains(got, `"^bench_sess-abc_mysql_cdc_connect$"`) {
+		t.Errorf("expected anchored Connect topic regex for mysql_cdc; got:\n%s", got)
+	}
+}
+
 func TestCombineReset_EmptySteps(t *testing.T) {
 	got, err := combineReset("postgres_cdc", nil, map[string]string{})
 	if err != nil {
diff --git a/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml b/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml
new file mode 100644
index 0000000000..434b688eb2
--- /dev/null
+++ b/benchmarking/aws/scenarios/mysql/orders-soak-pr.yaml
@@ -0,0 +1,80 @@
+name: mysql-orders-soak-pr
+description: |
+  PR before/after soak: the same sustained-moderate mysql_cdc load as
+  mysql-orders-soak, run twice on the SAME infrastructure — once with the
+  PR's base build, once with the PR's build — so the only variable is the
+  binary. 30 minutes per arm keeps the whole /soak round under ~2.5h while
+  still long enough for leak slopes and stalls to separate from noise.
+
+  The scenario name deliberately differs from the nightly's: CloudWatch
+  metrics land in a separate series that no alarm matches, and the runner
+  skips both the rolling-baseline comparator and the soak-index entry for
+  binary-arm runs — a deliberate A/B is not a nightly baseline sample.
+
+connector: mysql_cdc
+stack: mysql
+soak: true
+
+infra:
+  source:
+    instance_class: db.r6g.xlarge
+    # 400 GB gp3 floor: below it RDS forbids iops/storage_throughput, and
+    # 12000/500 are the >=400 GB tier's free baseline (see orders-soak.yaml).
+    storage_gb: 400
+    iops: 12000
+    storage_throughput: 500
+    parameters:
+      binlog_format: "ROW"
+      binlog_row_image: "FULL"
+      binlog_checksum: "NONE"
+  runner:
+    instance_type: c8g.xlarge   # 4 vCPU: reservedCores(2) + cpu_points[2] == 4
+
+dataset:
+  initial_rows: 0
+  row_size_bytes: 1200
+  tables: [orders]
+  seeder: cdc-rows-mysql
+
+workload:
+  write_rate_per_sec: 10000
+  duration: 30m
+  warmup: 5m
+
+pipeline:
+  cache_resources:
+    - label: bench_checkpoint
+      memory: {}
+  input:
+    mysql_cdc:
+      flavor: mysql
+      dsn: ${MYSQL_DSN}
+      tls:
+        skip_cert_verify: true   # RDS-internal CA isn't in the runner image
+      stream_snapshot: false
+      tables: [orders]
+      checkpoint_cache: bench_checkpoint
+      checkpoint_key: mysql_bench
+      # >= batching.count, or the default (1024) caps the pipeline at one
+      # batch in flight — see orders-soak.yaml.
+      checkpoint_limit: 10000
+      batching:
+        count: 5000
+        period: 1s
+
+matrix:
+  cpu_points: [2]
+  # Arms differ ONLY by binary — the runner enforces this for soak A/Bs.
+  # The workflow supplies the paths via --binary base=... --binary pr=...
+  arms:
+    - id: base
+      binary: base
+    - id: pr
+      binary: pr
+
+reset:
+  # Binlog retention first (RDS purges backed-up binlogs without it — see
+  # orders-soak.yaml), then a per-arm TRUNCATE so both arms start from the
+  # same bounded table; no slot/publication analogue exists for mysql.
+  - sql: "CALL mysql.rds_set_configuration('binlog retention hours', 24)"
+  - sql: "TRUNCATE TABLE orders"
diff --git a/benchmarking/aws/scenarios/mysql/orders-soak.yaml b/benchmarking/aws/scenarios/mysql/orders-soak.yaml
new file mode 100644
index 0000000000..d31fed1e85
--- /dev/null
+++ b/benchmarking/aws/scenarios/mysql/orders-soak.yaml
@@ -0,0 +1,96 @@
+name: mysql-orders-soak
+description: |
+  Soak: hold Connect's mysql_cdc at a fixed, sustained-moderate load (10K
+  writes/sec ≈ 12 MB/s — roughly 10-15% of the ~100-115 MB/s read ceiling
+  measured for mysql_cdc, see docs history and the benchmarking branch's
+  mysql-orders-cdc sweep) for 90 minutes at a single vCPU point. Same
+  purpose as postgres-orders-soak: catch leaks, stalls, and rotation bugs
+  that a short max-load sweep never runs long enough to hit.
+
+  Password auth for now. The mysql soak is the designated future home of
+  the ~15-min IAM credential-rotation lever (RDS IAM tokens ride
+  normal-protocol MySQL connections, unlike postgres replication — see
+  postgres-orders-soak's description and #4258/#4668), but that arrives as
+  its own increment: IAM needs the AWSAuthenticationPlugin DB user created
+  in reset, the rds-db:connect policy in the stack, and a live validation
+  pass. Everything here works unchanged when that lands — only the DSN,
+  the input's aws block, and reset steps change.
+
+connector: mysql_cdc
+stack: mysql
+soak: true
+
+infra:
+  source:
+    instance_class: db.r6g.xlarge
+    # Same RDS gp3 tiering as postgres: below 400 GB you may not set
+    # iops/storage_throughput at all; at >=400 GB the free baseline is
+    # 12000 / 500 and those are the legal minimums (live-confirmed on the
+    # postgres stack 2026-08-12).
+    storage_gb: 400
+    iops: 12000
+    storage_throughput: 500
+    # Explicit for self-documentation; identical to the stack defaults.
+    # ROW + FULL are required by mysql_cdc, NONE keeps go-mysql compatible
+    # across server versions.
+    parameters:
+      binlog_format: "ROW"
+      binlog_row_image: "FULL"
+      binlog_checksum: "NONE"
+  runner:
+    instance_type: c8g.xlarge   # 4 vCPU: reservedCores(2) + cpu_points[2] == 4
+
+dataset:
+  initial_rows: 0          # reset TRUNCATEs; seed still CREATEs the table
+  row_size_bytes: 1200
+  tables: [orders]
+  seeder: cdc-rows-mysql
+
+workload:
+  write_rate_per_sec: 10000
+  duration: 90m
+  warmup: 5m
+
+pipeline:
+  # mysql_cdc requires a cache resource for its binlog-position checkpoint.
+  # Memory is deliberate: the soak measures one uninterrupted process, and
+  # with stream_snapshot: false a (hypothetical) restart resumes from the
+  # current binlog position anyway.
+  cache_resources:
+    - label: bench_checkpoint
+      memory: {}
+  input:
+    mysql_cdc:
+      flavor: mysql
+      dsn: ${MYSQL_DSN}
+      tls:
+        skip_cert_verify: true   # RDS-internal CA isn't in the runner image; mysql_cdc uses NewTLSField (no `enabled` toggle)
+      stream_snapshot: false
+      tables: [orders]
+      checkpoint_cache: bench_checkpoint
+      checkpoint_key: mysql_bench
+      # Must be >= batching.count: the default (1024) admits an oversized
+      # batch only when nothing is pending, which serializes the pipeline to
+      # exactly one 5000-row batch in flight per produce+ack round trip.
+      checkpoint_limit: 10000
+      batching:
+        count: 5000
+        period: 1s
+
+matrix:
+  cpu_points: [2]
+
+reset:
+  # RDS purges binlog files as soon as they're backed up unless this is
+  # set — backup_retention_period=1 in the module only ENABLES binlog (see
+  # modules/rds-mysql). Without it, any mid-soak reconnect that outlives
+  # the purge finds its checkpointed position gone and dies with ERROR
+  # 1236 — an infra artifact indistinguishable from a connector bug. 24h
+  # comfortably outlives the longest run; the setting is per-instance and
+  # each session provisions a fresh one, so it must be set every run.
+  - sql: "CALL mysql.rds_set_configuration('binlog retention hours', 24)"
+  # TRUNCATE keeps the table bounded across runs on reused infra. Unlike
+  # postgres there is no slot/publication state to clear: with
+  # stream_snapshot: false and a fresh in-memory checkpoint cache, the
+  # connector always starts from the current binlog position.
+  - sql: "TRUNCATE TABLE orders"
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/main.go b/benchmarking/aws/seeders/cdc-rows-mysql/main.go
new file mode 100644
index 0000000000..767c18e1a3
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/main.go
@@ -0,0 +1,52 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Fprintln(os.Stderr, "usage: cdc-rows-mysql {seed|workload} [flags]")
+		os.Exit(2)
+	}
+	cmd := os.Args[1]
+	switch cmd {
+	case "seed":
+		fs := flag.NewFlagSet("seed", flag.ExitOnError)
+		tables := fs.String("tables", "orders", "comma-separated table list")
+		rows := fs.Int64("rows", 1_000_000, "rows per table")
+		rowSize := fs.Int("row-size", 1200, "approximate row size in bytes")
+		_ = fs.Parse(os.Args[2:])
+		if err := seed(context.Background(), strings.Split(*tables, ","), *rows, *rowSize); err != nil {
+			fmt.Fprintln(os.Stderr, "seed:", err)
+			os.Exit(1)
+		}
+	case "workload":
+		fs := flag.NewFlagSet("workload", flag.ExitOnError)
+		tables := fs.String("tables", "orders", "comma-separated table list")
+		rowSize := fs.Int("row-size", 1200, "approximate row size in bytes")
+		rate := fs.Int("rate", 5000, "writes per second total across tables")
+		dur := fs.Duration("duration", 15*time.Minute, "total duration")
+		_ = fs.Parse(os.Args[2:])
+		if err := workload(context.Background(), strings.Split(*tables, ","), *rowSize, *rate, *dur); err != nil {
+			fmt.Fprintln(os.Stderr, "workload:", err)
+			os.Exit(1)
+		}
+	default:
+		fmt.Fprintln(os.Stderr, "unknown subcommand:", cmd)
+		os.Exit(2)
+	}
+}
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/sql.go b/benchmarking/aws/seeders/cdc-rows-mysql/sql.go
new file mode 100644
index 0000000000..3039834f22
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/sql.go
@@ -0,0 +1,393 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+// This is the mysql port of cdc-rows-postgres: same worker/tick fairness
+// math (every row and every write/sec accounted for, no truncation), same
+// distinct-payload pool (identical payloads compress away and misreport
+// throughput by 11-17x — see payloadPoolSize), with database/sql +
+// go-sql-driver in place of pgx and `?` placeholders in place of `$n`.
+
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+func openDB(maxConns int) (*sql.DB, error) {
+	cfg, err := mysql.ParseDSN(os.Getenv("MYSQL_DSN"))
+	if err != nil {
+		return nil, err
+	}
+	// pgx (the postgres seeder's driver) caches prepared statements
+	// transparently; go-sql-driver does NOT — without InterpolateParams it
+	// returns driver.ErrSkip for parameterized Exec and database/sql wraps
+	// every insert in a hidden server-side Prepare+Execute+Close, doubling
+	// statement traffic and roughly halving the per-worker ceiling the
+	// 16-worker design assumes. Payloads are base64 ASCII, so client-side
+	// interpolation is injection-safe here.
+	cfg.InterpolateParams = true
+	connector, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, err
+	}
+	db := sql.OpenDB(connector)
+	db.SetMaxOpenConns(maxConns)
+	db.SetMaxIdleConns(maxConns)
+	return db, nil
+}
+
+func seed(ctx context.Context, tables []string, rows int64, rowSize int) error {
+	db, err := openDB(16)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	for _, table := range tables {
+		if err := ensureTable(ctx, db, table, rowSize); err != nil {
+			return err
+		}
+	}
+	var wg sync.WaitGroup
+	errCh := make(chan error, len(tables))
+	for _, table := range tables {
+		wg.Add(1)
+		go func(t string) {
+			defer wg.Done()
+			errCh <- bulkInsert(ctx, db, t, rows, rowSize)
+		}(table)
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ensureTable(ctx context.Context, db *sql.DB, table string, rowSize int) error {
+	stmts := []string{
+		"DROP TABLE IF EXISTS " + table,
+		fmt.Sprintf(`CREATE TABLE %s (
+			id          BIGINT AUTO_INCREMENT PRIMARY KEY,
+			created_at  DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+			payload     TEXT NOT NULL
+		) ENGINE=InnoDB`, table),
+	}
+	for _, s := range stmts {
+		if _, err := db.ExecContext(ctx, s); err != nil {
+			return fmt.Errorf("%s: %w", s, err)
+		}
+	}
+	_ = rowSize
+	return nil
+}
+
+// workerRowCounts splits rows across workers as evenly as possible,
+// handing the remainder (rows % workers) one extra row each to the first
+// `remainder` workers. Truncating division alone (rows/workers) silently
+// drops up to workers-1 rows off the total — for rows=10, workers=16 that
+// truncates to 0 per worker and seeds nothing at all — so every row must be
+// accounted for in the returned counts, which always sum to exactly rows.
+func workerRowCounts(rows int64, workers int) []int64 {
+	counts := make([]int64, workers)
+	if workers <= 0 {
+		return counts
+	}
+	base := rows / int64(workers)
+	remainder := rows % int64(workers)
+	for w := range workers {
+		counts[w] = base
+		if int64(w) < remainder {
+			counts[w]++
+		}
+	}
+	return counts
+}
+
+func bulkInsert(ctx context.Context, db *sql.DB, table string, rows int64, rowSize int) error {
+	const workers = 16
+	counts := workerRowCounts(rows, workers)
+	pool := randomPayloadPool(rowSize, payloadPoolSize)
+	start := time.Now()
+	var wg sync.WaitGroup
+	errCh := make(chan error, workers)
+	for w := range workers {
+		wg.Add(1)
+		workerRows := counts[w]
+		go func() {
+			defer wg.Done()
+			// A worker whose share rounded down to zero (e.g. rows=10 spread
+			// over 16 workers) has nothing to insert.
+			if workerRows == 0 {
+				errCh <- nil
+				return
+			}
+			const batchSize = 1000
+			conn, err := db.Conn(ctx)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			defer conn.Close()
+
+			cursor := 0
+			done := int64(0)
+			// The full-size statement is built once and reused for every
+			// full batch. Only the trailing partial batch — whatever is
+			// left after the last full batch — needs a statement sized to
+			// exactly that many rows, otherwise the loop either overshoots
+			// (inserting a full batch when fewer rows remain) or requires
+			// padding args that don't exist.
+			var fullStmt string
+			var fullArgs []any
+			if workerRows >= batchSize {
+				fullStmt = fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, valuesList(batchSize))
+				fullArgs = make([]any, batchSize)
+			}
+			for workerRows-done >= batchSize {
+				fillArgs(fullArgs, pool, &cursor)
+				if _, err := conn.ExecContext(ctx, fullStmt, fullArgs...); err != nil {
+					errCh <- err
+					return
+				}
+				done += batchSize
+			}
+			if remaining := workerRows - done; remaining > 0 {
+				stmt := fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, valuesList(int(remaining)))
+				args := make([]any, remaining)
+				fillArgs(args, pool, &cursor)
+				if _, err := conn.ExecContext(ctx, stmt, args...); err != nil {
+					errCh <- err
+					return
+				}
+			}
+			errCh <- nil
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			return err
+		}
+	}
+	fmt.Printf("seeded %d rows into %s in %s\n", rows, table, time.Since(start))
+	return nil
+}
+
+// ticksPerSecond is the number of 100ms ticker fires per second that each
+// workload worker uses to spread its per-second quota out smoothly instead
+// of bursting it all in one Exec.
+const ticksPerSecond = 10
+
+// perWorkerRate splits a total rows/sec rate across workers as evenly as
+// possible, handing the remainder (rate % workers) one extra row/sec each to
+// the first `remainder` workers. Truncating division alone drops up to
+// workers-1 rows/sec off the total, which is exactly the kind of small,
+// deterministic undershoot that lets a backlog metric drift forever on an
+// otherwise healthy run.
+func perWorkerRate(rate, workers int) []int {
+	rates := make([]int, workers)
+	if workers <= 0 {
+		return rates
+	}
+	base := rate / workers
+	remainder := rate % workers
+	for w := range workers {
+		rates[w] = base
+		if w < remainder {
+			rates[w]++
+		}
+	}
+	return rates
+}
+
+// tickCounts splits one worker's per-second quota across the ticksPerSecond
+// ticks of its 100ms ticker, again distributing the remainder so the ticks
+// sum to exactly ratePerWorker. This mirrors perWorkerRate one level down;
+// without it, a worker's per-tick truncation (ratePerWorker/10) compounds
+// with the per-worker truncation above into a rate that's measurably below
+// what was requested.
+func tickCounts(ratePerWorker int) [ticksPerSecond]int {
+	var counts [ticksPerSecond]int
+	base := ratePerWorker / ticksPerSecond
+	remainder := ratePerWorker % ticksPerSecond
+	for t := range ticksPerSecond {
+		counts[t] = base
+		if t < remainder {
+			counts[t]++
+		}
+	}
+	return counts
+}
+
+func workload(ctx context.Context, tables []string, rowSize, rate int, dur time.Duration) error {
+	// A single goroutine driving large per-tick batches caps around 30-40K
+	// inserts/sec on c8g.large because statement parsing + one network RTT
+	// per tick eats the budget. Spread across workers, each with a smaller
+	// batch, so the scenario's write_rate_per_sec is actually achievable.
+	// 16 workers handles 150K writes/sec comfortably (each worker ~9.4K/sec,
+	// well under the per-worker ceiling).
+	const workers = 16
+	db, err := openDB(workers)
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+
+	rates := perWorkerRate(rate, workers)
+	deadline := time.Now().Add(dur)
+	var wg sync.WaitGroup
+	errCh := make(chan error, workers)
+	for w := range workers {
+		wg.Add(1)
+		workerIdx := w
+		ratePerWorker := rates[w]
+		go func() {
+			defer wg.Done()
+			// Distinct payloads (built once) so change events aren't trivially
+			// compressible — see payloadPoolSize.
+			pool := randomPayloadPool(rowSize, payloadPoolSize)
+			cursor := 0
+			counts := tickCounts(ratePerWorker)
+
+			// Ticks only ever need one of two row counts (base or base+1),
+			// so the VALUES clauses are built once here instead of on every
+			// tick; a size of 0 is skipped since such a tick issues no Exec.
+			base := ratePerWorker / ticksPerSecond
+			remainder := ratePerWorker % ticksPerSecond
+			var baseValues, plusValues string
+			var baseArgs, plusArgs []any
+			if base > 0 {
+				baseValues = valuesList(base)
+				baseArgs = make([]any, base)
+			}
+			if remainder > 0 {
+				plusValues = valuesList(base + 1)
+				plusArgs = make([]any, base+1)
+			}
+
+			ticker := time.NewTicker(100 * time.Millisecond)
+			defer ticker.Stop()
+			tIdx := workerIdx
+			tickInSecond := 0
+			for {
+				select {
+				case <-ctx.Done():
+					errCh <- ctx.Err()
+					return
+				case <-ticker.C:
+					if time.Now().After(deadline) {
+						errCh <- nil
+						return
+					}
+					// The table rotates every tick regardless of whether
+					// this tick has any rows to insert, so the rotation
+					// stays in lockstep across workers.
+					table := tables[tIdx%len(tables)]
+					tIdx++
+					count := counts[tickInSecond]
+					tickInSecond = (tickInSecond + 1) % ticksPerSecond
+					if count == 0 {
+						continue
+					}
+					values, args := baseValues, baseArgs
+					if count == base+1 {
+						values, args = plusValues, plusArgs
+					}
+					stmt := fmt.Sprintf("INSERT INTO %s (created_at, payload) VALUES %s", table, values)
+					fillArgs(args, pool, &cursor)
+					if _, err := db.ExecContext(ctx, stmt, args...); err != nil {
+						errCh <- err
+						return
+					}
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+			return err
+		}
+	}
+	return nil
+}
+
+// payloadPoolSize is the number of distinct random payloads cycled per worker.
+//
+// Ported from cdc-rows-mongodb, the only seeder that originally had this right.
+// Reusing ONE identical payload for every row makes each producer batch
+// trivially compressible, and that alone accounted for the 11-17x gap between
+// Connect's self-reported throughput and the broker's byte counters across the
+// postgres, mysql, oracle and sqlserver benches. Mongo's calibration note is the
+// authority on the size: 4096 comfortably exceeds one compression batch, where
+// 1024 still left ~1.5x compressible.
+const payloadPoolSize = 4096
+
+// valuesList builds a multi-row VALUES clause with ONE PLACEHOLDER PER ROW:
+// (NOW(6),?),(NOW(6),?),...
+//
+// Distinct payloads need distinct placeholders (see cdc-rows-postgres, whose
+// single-repeated-placeholder bug this port never inherits). MySQL's
+// per-statement limit is max_allowed_packet, not a placeholder count; n=1000
+// rows of ~1.2 KB is ~1.2 MB, inside RDS's 64 MB default.
+func valuesList(n int) string {
+	var sb strings.Builder
+	for i := range n {
+		if i > 0 {
+			sb.WriteString(",")
+		}
+		sb.WriteString("(NOW(6),?)")
+	}
+	return sb.String()
+}
+
+// fillArgs refills args from the pool, advancing the cursor. Refilling every
+// batch matters: filling once would make each batch byte-identical to the last,
+// which compresses just as well as a single repeated payload did.
+func fillArgs(args []any, pool []string, cursor *int) {
+	for i := range args {
+		args[i] = pool[*cursor%len(pool)]
+		*cursor++
+	}
+}
+
+// randomPayloadPool builds n distinct random payloads of ~size bytes.
+func randomPayloadPool(size, n int) []string {
+	pool := make([]string, n)
+	for i := range pool {
+		pool[i] = randomPayload(size)
+	}
+	return pool
+}
+
+func randomPayload(size int) string {
+	b := make([]byte, (size*3)/4+1)
+	_, _ = rand.Read(b)
+	s := base64.StdEncoding.EncodeToString(b)
+	if len(s) > size {
+		s = s[:size]
+	}
+	return s
+}
diff --git a/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go b/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go
new file mode 100644
index 0000000000..2625d3c09c
--- /dev/null
+++ b/benchmarking/aws/seeders/cdc-rows-mysql/sql_test.go
@@ -0,0 +1,153 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerRowCounts(t *testing.T) {
+	tests := []struct {
+		name    string
+		rows    int64
+		workers int
+		want    int64 // expected sum, checked below alongside per-worker invariants
+	}{
+		{name: "evenly divisible", rows: 1_000_000, workers: 16, want: 1_000_000},
+		{name: "fewer rows than workers", rows: 10, workers: 16, want: 10},
+		{name: "single row", rows: 1, workers: 16, want: 1},
+		{name: "zero rows", rows: 0, workers: 16, want: 0},
+		{name: "not evenly divisible", rows: 1_000_001, workers: 16, want: 1_000_001},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			counts := workerRowCounts(tt.rows, tt.workers)
+			require.Len(t, counts, tt.workers)
+
+			var sum int64
+			for _, c := range counts {
+				require.GreaterOrEqual(t, c, int64(0))
+				sum += c
+			}
+			require.Equal(t, tt.want, sum)
+
+			// The spread must never exceed one row between the largest and
+			// smallest share, otherwise the remainder wasn't distributed
+			// evenly (e.g. it was dumped entirely on the first worker).
+			lo, hi := counts[0], counts[0]
+			for _, c := range counts {
+				if c < lo {
+					lo = c
+				}
+				if c > hi {
+					hi = c
+				}
+			}
+			require.LessOrEqual(t, hi-lo, int64(1))
+		})
+	}
+}
+
+func TestWorkerRowCounts_ExactShares(t *testing.T) {
+	// rows=10, workers=16: the first 10 workers get exactly 1 row each and
+	// the rest get 0 — this is the case that used to seed zero rows total.
+	counts := workerRowCounts(10, 16)
+	for w, c := range counts {
+		if w < 10 {
+			require.Equal(t, int64(1), c, "worker %d", w)
+		} else {
+			require.Equal(t, int64(0), c, "worker %d", w)
+		}
+	}
+}
+
+func TestPerWorkerRate(t *testing.T) {
+	tests := []struct {
+		name    string
+		rate    int
+		workers int
+		want    int
+	}{
+		{name: "declared soak rate", rate: 10_000, workers: 16, want: 10_000},
+		{name: "rate below worker count", rate: 3, workers: 16, want: 3},
+		{name: "large rate", rate: 160_000, workers: 16, want: 160_000},
+		{name: "zero rate", rate: 0, workers: 16, want: 0},
+		{name: "not evenly divisible", rate: 10_001, workers: 16, want: 10_001},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rates := perWorkerRate(tt.rate, tt.workers)
+			require.Len(t, rates, tt.workers)
+
+			var sum int
+			for _, r := range rates {
+				require.GreaterOrEqual(t, r, 0)
+				sum += r
+			}
+			require.Equal(t, tt.want, sum)
+		})
+	}
+}
+
+func TestTickCounts(t *testing.T) {
+	tests := []struct {
+		name          string
+		ratePerWorker int
+	}{
+		{name: "evenly divisible by ticks", ratePerWorker: 625},
+		{name: "less than one tick", ratePerWorker: 1},
+		{name: "zero", ratePerWorker: 0},
+		{name: "large per-worker rate", ratePerWorker: 10_000},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			counts := tickCounts(tt.ratePerWorker)
+			require.Len(t, counts, ticksPerSecond)
+
+			var sum int
+			for _, c := range counts {
+				require.GreaterOrEqual(t, c, 0)
+				sum += c
+			}
+			require.Equal(t, tt.ratePerWorker, sum)
+		})
+	}
+}
+
+// TestWorkloadRateEndToEnd pins the exact totals required by the soak
+// scenarios: distributing a declared rate across workers and then across
+// each worker's 10 ticks/sec must land on the declared rate exactly, with
+// no compounding truncation.
+func TestWorkloadRateEndToEnd(t *testing.T) {
+	tests := []struct {
+		name    string
+		rate    int
+		workers int
+	}{
+		{name: "shipped soak rate", rate: 10_000, workers: 16},
+		{name: "rate below worker count", rate: 3, workers: 16},
+		{name: "large rate", rate: 160_000, workers: 16},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rates := perWorkerRate(tt.rate, tt.workers)
+
+			var total int
+			for _, ratePerWorker := range rates {
+				counts := tickCounts(ratePerWorker)
+				for _, c := range counts {
+					total += c
+				}
+			}
+			require.Equal(t, tt.rate, total)
+		})
+	}
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/main.tf b/benchmarking/aws/terraform/modules/rds-mysql/main.tf
new file mode 100644
index 0000000000..814fe8bbce
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/main.tf
@@ -0,0 +1,85 @@
+resource "aws_db_subnet_group" "this" {
+  name       = "${var.name_prefix}-my"
+  subnet_ids = var.subnet_ids
+}
+
+resource "aws_security_group" "this" {
+  name        = "${var.name_prefix}-my-sg"
+  description = "Allow MySQL from bench clients"
+  vpc_id      = var.vpc_id
+
+  dynamic "ingress" {
+    for_each = var.client_sg_ids
+    content {
+      from_port       = 3306
+      to_port         = 3306
+      protocol        = "tcp"
+      security_groups = [ingress.value]
+    }
+  }
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+}
+
+resource "aws_db_parameter_group" "this" {
+  name   = "${var.name_prefix}-my"
+  family = "mysql8.0"
+  dynamic "parameter" {
+    for_each = var.parameters
+    content {
+      name         = parameter.key
+      value        = parameter.value
+      apply_method = "pending-reboot"
+    }
+  }
+}
+
+resource "random_password" "master" {
+  length  = 20
+  special = false
+}
+
+resource "aws_db_instance" "this" {
+  identifier             = "${var.name_prefix}-my"
+  engine                 = "mysql"
+  engine_version         = var.engine_version
+  instance_class         = var.instance_class
+  allocated_storage      = var.storage_gb
+  storage_type           = "gp3"
+  iops                   = var.iops
+  storage_throughput     = var.storage_throughput
+  db_name                = var.db_name
+  username               = var.master_username
+  password               = random_password.master.result
+  parameter_group_name   = aws_db_parameter_group.this.name
+  db_subnet_group_name   = aws_db_subnet_group.this.name
+  vpc_security_group_ids = [aws_security_group.this.id]
+  skip_final_snapshot    = true
+  deletion_protection    = false
+  publicly_accessible    = false
+  apply_immediately      = true
+
+  # CRITICAL: backup_retention_period > 0 is what ENABLES binlog on RDS
+  # MySQL at all — with backups off there is no binlog and mysql_cdc has
+  # nothing to read. It does NOT retain the binlog: RDS purges binlog files
+  # as soon as they're backed up unless 'binlog retention hours' is set,
+  # and that is a runtime stored procedure
+  # (CALL mysql.rds_set_configuration), not a parameter-group knob, so the
+  # scenarios' reset steps call it on every run (see scenarios/mysql/).
+  backup_retention_period = 1
+  # Pin the backup window to off-hours UTC. Without this, RDS picks a random
+  # daily slot; the 2026-05-21 smoke saw the 8 vCPU sweep point degrade from
+  # ~100 MB/s to ~60 MB/s mid-window because a backup overlapped (gp3
+  # throughput is shared between user writes and snapshot copy). 06:00-08:00
+  # UTC is well before US-Pacific working hours when benches typically run,
+  # and ends before the nightly soak's 08:10 UTC cron even fires — a nightly
+  # soak instance (created ~08:15, destroyed ~2.5h later) never lives through
+  # the window at all, only through the unavoidable on-creation snapshot,
+  # which lands during provisioning rather than the measured run.
+  backup_window = "06:00-08:00"
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf b/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf
new file mode 100644
index 0000000000..306c468d56
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/outputs.tf
@@ -0,0 +1,16 @@
+output "mysql_dsn" {
+  # go-sql-driver/mysql DSN: user:pass@tcp(host:port)/db?params
+  # parseTime=true maps DATETIME → time.Time at the driver layer.
+  # tls=skip-verify because the RDS-internal CA isn't in the runner image.
+  value     = "${var.master_username}:${random_password.master.result}@tcp(${aws_db_instance.this.address}:3306)/${var.db_name}?parseTime=true&tls=skip-verify"
+  sensitive = true
+}
+output "mysql_endpoint" { value = aws_db_instance.this.address }
+output "mysql_host" { value = aws_db_instance.this.address }
+output "mysql_port" { value = "3306" }
+output "mysql_user" { value = var.master_username }
+output "mysql_db" { value = var.db_name }
+output "mysql_password" {
+  value     = random_password.master.result
+  sensitive = true
+}
diff --git a/benchmarking/aws/terraform/modules/rds-mysql/variables.tf b/benchmarking/aws/terraform/modules/rds-mysql/variables.tf
new file mode 100644
index 0000000000..2693602353
--- /dev/null
+++ b/benchmarking/aws/terraform/modules/rds-mysql/variables.tf
@@ -0,0 +1,51 @@
+variable "name_prefix" { type = string }
+variable "vpc_id" { type = string }
+variable "subnet_ids" {
+  type = list(string)
+}
+variable "client_sg_ids" {
+  type        = list(string)
+  description = "SGs allowed to connect on 3306"
+}
+variable "instance_class" {
+  type    = string
+  default = "db.r6g.2xlarge"
+}
+variable "storage_gb" {
+  type    = number
+  default = 400
+}
+variable "iops" {
+  type    = number
+  default = 12000
+}
+variable "engine_version" {
+  type    = string
+  default = "8.0.46"
+}
+variable "db_name" {
+  type    = string
+  default = "benchdb"
+}
+variable "master_username" {
+  type    = string
+  default = "bench"
+}
+variable "parameters" {
+  type = map(string)
+  # binlog_format=ROW + binlog_row_image=FULL are required by mysql_cdc.
+  # binlog_checksum=NONE keeps the go-mysql client compatible across versions.
+  default = {
+    binlog_format    = "ROW"
+    binlog_row_image = "FULL"
+    binlog_checksum  = "NONE"
+  }
+}
+variable "storage_throughput" {
+  # gp3 throughput in MiB/s. null = RDS default. Same tiering as postgres:
+  # below 400 GB RDS forbids setting iops/storage_throughput at all; at
+  # >=400 GB the free baseline is 12000 IOPS / 500 MiB/s and those are the
+  # legal minimums (see modules/rds-postgres and the postgres soak scenario).
+  type    = number
+  default = null
+}
diff --git a/benchmarking/aws/terraform/persistent/variables.tf b/benchmarking/aws/terraform/persistent/variables.tf
index ae69b0e909..a5eea2c404 100644
--- a/benchmarking/aws/terraform/persistent/variables.tf
+++ b/benchmarking/aws/terraform/persistent/variables.tf
@@ -21,5 +21,9 @@ variable "soak_scenarios" {
       connector = "postgres_cdc"
       scenario  = "postgres-orders-soak"
     }
+    mysql = {
+      connector = "mysql_cdc"
+      scenario  = "mysql-orders-soak"
+    }
   }
 }
diff --git a/benchmarking/aws/terraform/stacks/mysql/main.tf b/benchmarking/aws/terraform/stacks/mysql/main.tf
new file mode 100644
index 0000000000..a0e5d2b2ee
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/main.tf
@@ -0,0 +1,54 @@
+terraform {
+  required_version = ">= 1.10" # S3-native state locking (use_lockfile in backend.hcl)
+  required_providers {
+    aws    = { source = "hashicorp/aws", version = "~> 5.70" }
+    random = { source = "hashicorp/random", version = "~> 3.6" }
+  }
+  backend "s3" {}
+}
+
+provider "aws" {
+  region = var.region
+  default_tags {
+    tags = {
+      Project   = "redpanda-connect-bench"
+      Stack     = "mysql"
+      ManagedBy = "terraform"
+      # The cleanup lambda derives creation time for RDS subnet/parameter
+      # groups and security groups EXCLUSIVELY from this tag (see
+      # cleanup-lambda/sweep.go sessionCreatedAt) — those resources carry no
+      # creation-time field of their own. Without it, an aborted teardown
+      # leaves this stack's groups unsweepable forever, and the surviving
+      # security group blocks the shared VPC's deletion on every sweep.
+      "bench-session-id" = var.bench_session_id
+      # See shared/main.tf: exempt from the org cloud-nuke sweep; our own
+      # reaper (keyed on Project) still reaps this stack at the 4h TTL.
+      "cloud-nuke-excluded" = "true"
+    }
+  }
+}
+
+data "terraform_remote_state" "shared" {
+  backend = "s3"
+  config = {
+    bucket = "redpanda-connect-bench-tfstate"
+    region = var.region
+    key    = "shared/terraform.tfstate"
+  }
+}
+
+module "rds" {
+  source      = "../../modules/rds-mysql"
+  name_prefix = "rpcn-bench-my"
+  vpc_id      = data.terraform_remote_state.shared.outputs.vpc_id
+  subnet_ids  = data.terraform_remote_state.shared.outputs.private_subnet_ids
+  client_sg_ids = [
+    data.terraform_remote_state.shared.outputs.runner_sg_id,
+    data.terraform_remote_state.shared.outputs.load_gen_sg_id,
+  ]
+  instance_class     = var.instance_class
+  storage_gb         = var.storage_gb
+  iops               = var.iops
+  storage_throughput = var.storage_throughput
+  parameters         = var.parameters
+}
diff --git a/benchmarking/aws/terraform/stacks/mysql/outputs.tf b/benchmarking/aws/terraform/stacks/mysql/outputs.tf
new file mode 100644
index 0000000000..7984b53384
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/outputs.tf
@@ -0,0 +1,20 @@
+output "mysql_dsn" {
+  value     = module.rds.mysql_dsn
+  sensitive = true
+}
+output "mysql_endpoint" { value = module.rds.mysql_endpoint }
+# Discrete connection parts for the reset builder: the mysql CLI takes
+# -h/-P/-u/-p flags rather than a DSN URL (see combineReset in the runner).
+output "mysql_host" { value = module.rds.mysql_host }
+output "mysql_port" { value = module.rds.mysql_port }
+output "mysql_user" { value = module.rds.mysql_user }
+output "mysql_db" { value = module.rds.mysql_db }
+output "mysql_password" {
+  value     = module.rds.mysql_password
+  sensitive = true
+}
+# Exposed so scenarios can reference ${REGION} instead of hardcoding the
+# bench account's region (same contract as the postgres stack).
+output "region" {
+  value = var.region
+}
diff --git a/benchmarking/aws/terraform/stacks/mysql/variables.tf b/benchmarking/aws/terraform/stacks/mysql/variables.tf
new file mode 100644
index 0000000000..122a01b9a6
--- /dev/null
+++ b/benchmarking/aws/terraform/stacks/mysql/variables.tf
@@ -0,0 +1,34 @@
+variable "region" {
+  type    = string
+  default = "us-east-2"
+}
+variable "bench_session_id" {
+  # Runner-generated session ID ("bench-YYYYMMDD-HHMMSS", see newSessionID).
+  # Stamped on every resource via default_tags; the cleanup lambda decodes
+  # the embedded timestamp as the only age signal for resources whose
+  # Describe response carries no creation time. Same contract as the shared
+  # stack's variable of the same name, including the empty default: `runner
+  # down` rebuilds stack vars from the scenario's infra.source only, so a
+  # required variable would abort the destroy (-input=false) and strand paid
+  # infra until the reaper's TTL. The tag value doesn't matter during destroy.
+  type    = string
+  default = ""
+}
+variable "instance_class" { type = string }
+variable "storage_gb" { type = number }
+variable "iops" { type = number }
+variable "parameters" {
+  type = map(string)
+  # binlog_format=ROW + binlog_row_image=FULL are what mysql_cdc requires;
+  # binlog_checksum=NONE keeps go-mysql compatible across server versions.
+  default = {
+    binlog_format    = "ROW"
+    binlog_row_image = "FULL"
+    binlog_checksum  = "NONE"
+  }
+}
+variable "storage_throughput" {
+  # gp3 throughput MiB/s; null = RDS default. See modules/rds-mysql.
+  type    = number
+  default = null
+}

From c00b99223fba91543621aa2b6399b6a38bfb2975 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Tue, 1 Sep 2026 08:59:18 -0700
Subject: [PATCH 2/2] bench(soak): match the scenario allowlist against the
 whole string
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

grep -Eq evaluates per line, so a multi-line dispatch input (reachable via
the REST dispatch API) passed the allowlist on its first conforming line
and was then written verbatim into the line-oriented $GITHUB_OUTPUT —
where an embedded `scenarios=` assignment or heredoc could re-assign the
matrix and fan one dispatch into N paid soak jobs. Bash's =~ anchors $ to
end-of-string and no class in the pattern admits a newline, so multi-line
values now fail both the plan job's gate and the per-job re-check.
Verified: the review comment's newline and heredoc payloads pass the old
grep and are rejected by =~; valid paths still pass.

soak_pr.yml's sibling check is not affected: its `| xargs` flattens
newlines to spaces before the regex, and a space fails the allowlist.

Found by claude-review on #4750.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .github/workflows/soak_nightly.yml | 19 +++++++++++++++----
 1 file changed, 15 insertions(+), 4 deletions(-)

diff --git a/.github/workflows/soak_nightly.yml b/.github/workflows/soak_nightly.yml
index 1d905860cb..62b832be70 100644
--- a/.github/workflows/soak_nightly.yml
+++ b/.github/workflows/soak_nightly.yml
@@ -73,8 +73,16 @@ jobs:
             echo 'scenarios=["postgres/orders-soak", "mysql/orders-soak"]' >> "$GITHUB_OUTPUT"
             exit 0
           fi
-          if ! echo "${SCENARIO}" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'; then
-            echo "::error::Invalid scenario '${SCENARIO}' — expected a path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
+          # bash [[ =~ ]], NOT `echo | grep`: grep matches per line, so a
+          # multi-line value (reachable via the REST dispatch API) would pass
+          # on its first line and then write attacker-controlled extra lines
+          # into the line-oriented $GITHUB_OUTPUT below — re-assigning
+          # `scenarios` and fanning one dispatch into N paid jobs. Bash's =~
+          # anchors $ to end-of-STRING and no class here admits a newline,
+          # so any embedded newline fails the match.
+          re='^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'
+          if ! [[ "${SCENARIO}" =~ $re ]]; then
+            echo "::error::Invalid scenario — expected a single path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
             exit 1
           fi
           echo "scenarios=[\"${SCENARIO}\"]" >> "$GITHUB_OUTPUT"
@@ -110,8 +118,11 @@ jobs:
         env:
           SCENARIO: ${{ matrix.scenario }}
         run: |
-          if ! echo "${SCENARIO}" | grep -Eq '^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'; then
-            echo "::error::Invalid scenario '${SCENARIO}' — expected a path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
+          # Full-string match, same as the plan job (grep would pass a
+          # multi-line value on its first matching line).
+          re='^[A-Za-z0-9][A-Za-z0-9_-]*(/[A-Za-z0-9][A-Za-z0-9_-]*)*$'
+          if ! [[ "${SCENARIO}" =~ $re ]]; then
+            echo "::error::Invalid scenario — expected a single path of [A-Za-z0-9_-] segments, e.g. postgres/orders-soak"
             exit 1
           fi