Distributed Testing, Remote Engines, Network Topology, and Synchronization: Guided Hands-On Workflow

1. Safety envelope

2. Create the local target

Save fixtures/distributed_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse, parse_qs
import argparse
import json
import re
import threading
import time

FIXTURE_VERSION = "prompt25-distributed-fixture-v1"
SAFE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")

lock = threading.Lock()
event_log = None
state = {
    "requests": 0,
    "errors": 0,
    "duplicates": 0,
    "wrong_shard": 0,
    "accounts": set(),
    "by_engine": {},
}

def now_ms():
    return int(time.time() * 1000)

def snapshot():
    with lock:
        return {
            "requests": state["requests"],
            "errors": state["errors"],
            "duplicates": state["duplicates"],
            "wrong_shard": state["wrong_shard"],
            "accounts": sorted(state["accounts"]),
            "by_engine": dict(state["by_engine"]),
        }

def reset():
    with lock:
        state["requests"] = 0
        state["errors"] = 0
        state["duplicates"] = 0
        state["wrong_shard"] = 0
        state["accounts"].clear()
        state["by_engine"].clear()

def write_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()
        self.wfile.write(raw)

    def do_POST(self):
        if urlparse(self.path).path != "/reset":
            self.send_json(404, {"status": "not_found"})
            return
        reset()
        self.send_json(200, {"status": "reset", "epoch_ms": now_ms()})

    def do_GET(self):
        started = now_ms()
        parsed = urlparse(self.path)

        if parsed.path == "/health":
            self.send_json(200, {
                "status": "ok",
                "fixture_version": FIXTURE_VERSION,
                "epoch_ms": now_ms(),
            })
            return

        if parsed.path == "/stats":
            self.send_json(200, {
                "fixture_version": FIXTURE_VERSION,
                "epoch_ms": now_ms(),
                "state": snapshot(),
            })
            return

        if parsed.path != "/work":
            self.send_json(404, {"status": "not_found"})
            return

        q = parse_qs(parsed.query)
        run_id = q.get("run_id", [""])[0]
        engine = q.get("engine", [""])[0]
        account = q.get("account", [""])[0]
        thread_id = q.get("thread", [""])[0]

        if not all(SAFE.fullmatch(x or "") for x in (run_id, engine, account, thread_id)):
            self.send_json(400, {"status": "invalid_metadata"})
            return

        expected_prefix = {"engine-a": "a", "engine-b": "b"}.get(engine)
        if expected_prefix is None:
            self.send_json(400, {"status": "invalid_engine"})
            return

        status = 200
        reason = "ok"
        with lock:
            state["requests"] += 1
            state["by_engine"][engine] = state["by_engine"].get(engine, 0) + 1

            if not account.startswith(expected_prefix):
                status = 409
                reason = "wrong_shard"
                state["wrong_shard"] += 1
                state["errors"] += 1
            elif account in state["accounts"]:
                status = 409
                reason = "duplicate_account"
                state["duplicates"] += 1
                state["errors"] += 1
            else:
                state["accounts"].add(account)

        # Small deterministic service time; target is not the lab bottleneck.
        time.sleep(0.020)
        payload = {
            "status": reason,
            "run_id": run_id,
            "engine": engine,
            "account": account,
            "thread": thread_id,
        }
        self.send_json(status, payload)

        write_event({
            "ts_ms": now_ms(),
            "operation": "work",
            "status": status,
            "reason": reason,
            "run_id": run_id,
            "engine": engine,
            "account": account,
            "thread": thread_id,
            "service_wall_ms": now_ms() - started,
        })

    def log_message(self, format, *args):
        return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8025)
    parser.add_argument("--log", default="results/server-events.jsonl")
    args = parser.parse_args()

    global event_log
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    print(f"fixture_version={FIXTURE_VERSION}")
    print(f"listen=http://{args.host}:{args.port}")
    print(f"event_log={event_log}")
    ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()

if __name__ == "__main__":
    main()

Start:

python .\fixtures\distributed_fixture.py `
  --host 127.0.0.1 `
  --port 8025 `
  --log .\results\server-events.jsonl

Reset before each controlled run:

curl -X POST http://127.0.0.1:8025/reset

3. Create explicit six-row data shards

data/engine-a.csv:

ACCOUNT_ID
a001
a002
a003
a004
a005
a006

data/engine-b.csv:

ACCOUNT_ID
b001
b002
b003
b004
b005
b006

Each engine needs exactly six records because it will execute 2×3 samples. No recycling is allowed, so a missing/short shard stops threads rather than reusing identities.

4. Build a file/plugin manifest before deployment

Save tools/build_manifest.py:

import hashlib
import json
import sys
from pathlib import Path

if len(sys.argv) < 2:
    raise SystemExit("usage: build_manifest.py <file> [file ...]")

items = []
for raw in sys.argv[1:]:
    path = Path(raw)
    h = hashlib.sha256(path.read_bytes()).hexdigest()
    items.append({
        "path": str(path.resolve()),
        "bytes": path.stat().st_size,
        "sha256": h,
    })

print(json.dumps({"files": items}, indent=2))

python tools/build_manifest.py   plans/distributed-local.jmx   data/engine-a.csv   data/engine-b.csv

Also record jmeter -v, java -version and the names/hashes of any third-party JARs under lib/ext. Mandatory lab expected plugin set = none.

5. Separate global remote properties from controller result properties

config/remote-global.properties:

# Sent to every remote engine with -G<file> or individual -Gname=value flags.
threads=2
loops=3
target.host=127.0.0.1
target.port=8025
run.id=p25-remote
pacing.ms=50
connect.timeout.ms=500
response.timeout.ms=2000

config/remote-results.properties:

# Controller-side result contract.
jmeter.save.saveservice.output_format=csv
jmeter.save.saveservice.print_field_names=true
jmeter.save.saveservice.response_data=false
jmeter.save.saveservice.response_data.on_error=false
jmeter.save.saveservice.samplerData=false
jmeter.save.saveservice.responseHeaders=false
jmeter.save.saveservice.requestHeaders=false
jmeter.save.saveservice.hostname=true

# Save these engine-side variables into returned samples.
# Current JMeter propagates the requested sample variable names to remote servers.
sample_variables=ENGINE_ID,ACCOUNT_ID

# Explicit controller callback ports: 5000-5002 may be used.
client.rmi.localport=5000

# Keep the documented default sample sender explicit for the lab.
mode=StrippedBatch

# Fail the remote start if any configured engine cannot initialize.
client.continue_on_fail=false

-Gconfig/remote-global.properties sends the global workload/target properties to both engines. -q remote-results.properties configures the controller process's result/callback/sample-sender behavior. Engine-specific engine.id/data.file are intentionally absent from the global file.

6. Build the JMX in GUI

Test Plan
└── Thread Group — ${__P(threads,1)} threads × ${__P(loops,1)} loops
    ├── CSV Data Set Config — Engine shard
    │   Filename=${__P(data.file)}
    │   Variable Names=ACCOUNT_ID
    │   Recycle on EOF=false
    │   Stop thread on EOF=true
    │   Sharing mode=All threads
    ├── JSR223 PreProcessor — identify engine
    │   vars.put("ENGINE_ID", props.get("engine.id") ?: "unknown")
    └── HTTP Request — Work
        GET /work
          run_id=${__P(run.id,p25-local)}
          engine=${ENGINE_ID}
          account=${ACCOUNT_ID}
          thread=T${__threadNum}
        ├── Constant Timer ${__P(pacing.ms,50)} ms
        └── Response Assertion: response code is 200

The JSR223 PreProcessor uses built-in Groovy only; no plugin. It turns the engine-local engine.id property into thread-local variable ENGINE_ID. The CSV Data Set populates ACCOUNT_ID. Both are requested as sample variables so the controller aggregate CSV can count engines/accounts independently.

7. Create a disposable RMI SSL keystore

From the JMeter bin directory:

Windows:

Set-Location "$env:JMETER_HOME\bin"
.\create-rmi-keystore.bat

Unix-like:

cd "$JMETER_HOME/bin"
./create-rmi-keystore.sh

For the disposable lab, follow the documented default alias rmi; the generated certificate is short-lived (seven days) and the default passphrase is changeit. Copy/reference the resulting rmi_keystore.jks on controller and engines. Do not use the default password/certificate as a production remote-testing credential.

8. Start engine A with engine-local state

PowerShell, same-host loopback lab:

$Root = (Resolve-Path ".").Path
$J = "$env:JMETER_HOME\bin\jmeter.bat"

& $J -s `
  -Djava.rmi.server.hostname=127.0.0.1 `
  -Jserver_port=1099 `
  -Jserver.rmi.localport=4000 `
  -Jengine.id=engine-a `
  -Jdata.file="$Root\data\engine-a.csv" `
  -j "$Root\results\engine-a-jmeter.log"

State changed: registry/server RMI listeners open for engine A; engine-local properties identify shard A; no target traffic occurs until the controller starts a test.

9. Start engine B with different ports/shard

$Root = (Resolve-Path ".").Path
$J = "$env:JMETER_HOME\bin\jmeter.bat"

& $J -s `
  -Djava.rmi.server.hostname=127.0.0.1 `
  -Jserver_port=1100 `
  -Jserver.rmi.localport=4100 `
  -Jengine.id=engine-b `
  -Jdata.file="$Root\data\engine-b.csv" `
  -j "$Root\results\engine-b-jmeter.log"

On separate private hosts, use private addresses and the same fixed-port/SSL/version/file contract; do not advertise/listen on an uncontrolled public interface.

10. Verify engine state before load

Get-NetTCPConnection -State Listen |
  Where-Object LocalPort -in 1099,1100,4000,4100

Select-String `
  -Path .\results\engine-a-jmeter.log,.\results\engine-b-jmeter.log `
  -Pattern "RemoteJMeterEngine|RMI|ERROR|FATAL"

curl.exe --fail --silent http://127.0.0.1:8025/stats

For separate hosts, run jmeter -v/java -version/hash inventory on each engine and compare before remote start.

11. Write the workload prediction before -R

engine-a: 2 threads × 3 loops × 1 Work = 6 samples, accounts a001-a006
engine-b: 2 threads × 3 loops × 1 Work = 6 samples, accounts b001-b006

TOTAL configured samples = 6 + 6 = 12
TOTAL configured active-user threads at peak = 2 + 2 = 4
Expected target failures = 0
Expected unique accounts = 12

12. Run the remote test explicitly with -R/-G

PowerShell:

$Root = (Resolve-Path ".").Path
New-Item -ItemType Directory -Force .\results\p25-remote | Out-Null

& "$env:JMETER_HOME\bin\jmeter.bat" `
  -n `
  -t "$Root\plans\distributed-local.jmx" `
  -q "$Root\config\remote-results.properties" `
  -R127.0.0.1:1099,127.0.0.1:1100 `
  -G"$Root\config\remote-global.properties" `
  -l "$Root\results\p25-remote\aggregate.jtl" `
  -j "$Root\results\p25-remote\controller-jmeter.log"

-R is explicit and overrides remote_hosts. -r would instead run all hosts listed in remote_hosts. -G sends workload/target properties to each server. It does not copy the CSV files and does not create distinct engine IDs.

13. Verify aggregate/per-engine evidence

Save tools/verify_distributed.py:

import csv
import json
import sys
from collections import Counter
from pathlib import Path

if len(sys.argv) != 5:
    raise SystemExit(
        "usage: verify_distributed.py <aggregate.jtl> <server-events.jsonl> <run_id> <expected_per_engine>"
    )

jtl_path = Path(sys.argv[1])
events_path = Path(sys.argv[2])
run_id = sys.argv[3]
expected = int(sys.argv[4])

rows = list(csv.DictReader(jtl_path.open(newline="", encoding="utf-8")))
required = {"label", "success", "responseCode", "ENGINE_ID", "ACCOUNT_ID"}
missing = required - set(rows[0].keys() if rows else [])
if missing:
    raise SystemExit(f"JTL missing required columns: {sorted(missing)}")

by_engine = Counter(r["ENGINE_ID"] for r in rows)
accounts = [r["ACCOUNT_ID"] for r in rows]
failures = [r for r in rows if r["success"].lower() != "true"]

events = [
    json.loads(line)
    for line in events_path.read_text(encoding="utf-8").splitlines()
    if line.strip()
]
target = [
    e for e in events
    if e.get("operation") == "work" and e.get("run_id") == run_id
]
target_by_engine = Counter(e.get("engine") for e in target)
target_failures = [e for e in target if int(e.get("status", 0)) >= 400]

result = {
    "run_id": run_id,
    "expected_per_engine": expected,
    "expected_total": expected * 2,
    "jtl_rows": len(rows),
    "jtl_by_engine": dict(by_engine),
    "jtl_failures": len(failures),
    "jtl_unique_accounts": len(set(accounts)),
    "target_events": len(target),
    "target_by_engine": dict(target_by_engine),
    "target_failures": len(target_failures),
    "target_reasons": dict(Counter(e.get("reason") for e in target)),
}

ok = True
if len(rows) != expected * 2:
    ok = False
if by_engine != Counter({"engine-a": expected, "engine-b": expected}):
    ok = False
if len(set(accounts)) != expected * 2:
    ok = False
if target_by_engine != Counter({"engine-a": expected, "engine-b": expected}):
    ok = False
if failures or target_failures:
    ok = False

result["status"] = "PASS" if ok else "FAIL"
print(json.dumps(result, indent=2))
raise SystemExit(0 if ok else 3)

python tools/verify_distributed.py   results/p25-remote/aggregate.jtl   results/server-events.jsonl   p25-remote   6

Expected PASS: 12 aggregate rows, six per engine, 12 unique accounts, no JTL failures; target JSONL also shows six events per engine and no wrong-shard/duplicate failures.

14. Inspect all three JMeter logs

Controller:

Select-String .\results\p25-remote\controller-jmeter.log `
  -Pattern "ERROR|FATAL|remote|Remote|StrippedBatch"

Engines:

Select-String .\results\engine-*-jmeter.log `
  -Pattern "ERROR|FATAL|Starting|Finished|RMI"

Record controller/engine CPU/heap/network during the run even though it is tiny. A successful 12-row JTL does not prove large-scale generator capacity.

15. Understand -r without changing the lab

Equivalent host selection can be configured as:

# Controller property file:
remote_hosts=127.0.0.1:1099,127.0.0.1:1100

# Then:
jmeter -n -t plans/distributed-local.jmx -r ...

Prefer -R in one-off automation where an explicit run manifest should show the exact engine inventory; prefer -r where a controlled controller configuration owns the inventory.

16. Faithful independent-worker fallback

If local RMI SSL cannot be configured (restricted workstation/firewall), you can still validate workload math/data sharding by launching two independent CLI processes—this does not validate RMI/result transport:

# Engine A worker
jmeter -n -t plans/distributed-local.jmx `
  -Jthreads=2 -Jloops=3 -Jengine.id=engine-a `
  -Jdata.file="$Root\data\engine-a.csv" `
  -Jtarget.host=127.0.0.1 -Jtarget.port=8025 -Jrun.id=p25-sim `
  -Jsample_variables=ENGINE_ID,ACCOUNT_ID `
  -l results\sim-a.jtl -j results\sim-a.log

# Engine B worker: same, but engine-b / engine-b.csv -> sim-b.jtl

Merge the two JTLs:

import csv
import sys
from pathlib import Path

if len(sys.argv) < 4:
    raise SystemExit("usage: merge_jtl.py <out.csv> <engine-a.csv> <engine-b.csv> [...]")

out_path = Path(sys.argv[1])
inputs = [Path(p) for p in sys.argv[2:]]

all_rows = []
fieldnames = None
for path in inputs:
    with path.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        if fieldnames is None:
            fieldnames = reader.fieldnames
        elif reader.fieldnames != fieldnames:
            raise SystemExit(f"header mismatch: {path}")
        all_rows.extend(reader)

out_path.parent.mkdir(parents=True, exist_ok=True)
with out_path.open("w", newline="", encoding="utf-8") as handle:
    writer = csv.DictWriter(handle, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(sorted(all_rows, key=lambda r: int(r["timeStamp"])))

print(f"merged_rows={len(all_rows)}")

python tools/merge_jtl.py results/sim-aggregate.jtl results/sim-a.jtl results/sim-b.jtl

Then run the same verifier. This faithfully tests full-plan replication/data uniqueness/count aggregation, but it cannot prove RMI SSL, callback ports or controller sample-transfer capacity.

17. Challenge

You need exactly 120 concurrent threads total across three identical remote engines. What should the JMX Thread Group contain if every engine receives the same property?

40 threads per engine, not 120. Send -Gthreads=40 to all three engines, then verify aggregate achieved concurrency/counts and generator headroom. If engines have unequal capacity, use engine-specific worker designs/properties rather than pretending JMeter automatically divides 120.

Knowledge check

Official references and version notes

• JMeter User Manual — Remote (Distributed) Testing — full-plan replication, same-version guidance, data-file behavior, RMI SSL, remote ports, CLI remote execution, and sample sender modes.
• JMeter Getting Started — -r, -R, -G, -X, CLI/server mode, property semantics.
• JMeter Properties Reference — remote hosts, controller/server RMI ports, SSL keystore/truststore settings, client failure policy and result properties.
• Component Reference — CSV Data Set Config — distributed CSV file placement and relative/absolute path behavior.
• JMeter Listeners / Result files — CSV result fields, sample variables and host attribution.
• Apache JMeter downloads — current stable release and Java requirement.