Repository: nats-io/nats-server
Pull request #4821 (2023-11-27 12:47:38)

[ADDED] MQTT: test/bench using external client
This PR adds a number of MQTT-specific benchmarks, executed with an external test client (see https://github.com/ConnectEverything/mqtt-test/pull/1). This came out of my work on MQTT message delivery issues in 2.10, and is now packaged as a `go test --bench`, and added to CI.

The test client uses the Eclipse `paho` library, and can be executed standalone, pointed at an MQTT server. It measures the timing of its operations, and outputs it as JSON of reportable metrics.

The benchmarks here run the client, and report back the metrics it outputs. Memory allocations are measured server-only. 

Github action `benchmark` is used to compare the benchmark results to the base ref of the PR, or the prior result in a branch. If it fails a 120% threshold, the failure is reported as a comment on the commit(!), and the build fails.

To measure the effect of parallel-fetching retained messages (`SUBRET`), added `disable_retained_message_cache` MQTT option to the server. 

The following benchmarks are run, covering standalone/cluster, QOS, message sizes to 100K, as well as concurrent clients.
- `PUB` measures just publishing
- `PUBRET` measures publishing with the retained flag on
- `PUBSUB` measures message delivery times from publisher to subscriber
- `SUBRET` measures delivery of 100* retained messages to new subscriptions

Example of a failure (I rolled back the retained message "fanout" commit):
<img width="876" alt="image" src="https://github.com/nats-io/nats-server/assets/1187448/c56eb408-4e4a-4562-ab41-9c024b09ba47">
if you click on the comment link, you get:
<img width="1025" alt="image" src="https://github.com/nats-io/nats-server/assets/1187448/2b677bb5-8445-4ed0-af5b-bf8519d42f9f">

###### TODO
- `CON` measure CONNECT to CONNACK, with stored sessions
- `SUB`  measuring SUBSCRIBE to SUBACK


levb: @derekcollison the option is used only for testing, and is exposed because we use config files in the test utility functions  to initialize servers/clusters.

I could move the flag into the `(server.)srvMQTT` struct, set it after the server(s) have been started, would that be preferable? If there's a better mechanism of passing "test options" into a server, please advise.
derekcollison: Let's see what @kozlovic has to say, he has done a few of these testing flags / switches for things.

Patch:
diff --git a/server/mqtt.go b/server/mqtt.go
index 793d57d2..f64c160e 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -234,6 +234,8 @@ type mqttSessionManager struct {
 	sessions map[string]*mqttAccountSessionManager // key is account name
 }
 
+var testDisableRMSCache = false
+
 type mqttAccountSessionManager struct {
 	mu         sync.RWMutex
 	sessions   map[string]*mqttSession        // key is MQTT client ID
@@ -243,7 +245,7 @@ type mqttAccountSessionManager struct {
 	flapTimer  *time.Timer                    // Timer to perform some cleanup of the flappers map
 	sl         *Sublist                       // sublist allowing to find retained messages for given subscription
 	retmsgs    map[string]*mqttRetainedMsgRef // retained messages
-	rmsCache   sync.Map                       // map[subject]*mqttRetainedMsg
+	rmsCache   *sync.Map                      // map[subject]mqttRetainedMsg
 	jsa        mqttJSA
 	rrmLastSeq uint64        // Restore retained messages expected last sequence
 	rrmDoneCh  chan struct{} // To notify the caller that all retained messages have been loaded
@@ -1142,6 +1144,9 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 			quitCh: quitCh,
 		},
 	}
+	if !testDisableRMSCache {
+		as.rmsCache = &sync.Map{}
+	}
 	// TODO record domain name in as here
 
 	// The domain to communicate with may be required for JS calls.
@@ -1236,10 +1241,12 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	})
 
 	// Start the go routine that will clean up cached retained messages that expired.
-	s.startGoRoutine(func() {
-		defer s.grWG.Done()
-		as.cleanupRetainedMessageCache(s, closeCh)
-	})
+	if as.rmsCache != nil {
+		s.startGoRoutine(func() {
+			defer s.grWG.Done()
+			as.cleanupRetainedMessageCache(s, closeCh)
+		})
+	}
 
 	lookupStream := func(stream, txt string) (*StreamInfo, error) {
 		si, err := jsa.lookupStream(stream)
@@ -2243,9 +2250,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rf *mqttRetai
 
 			// Update the in-memory retained message cache but only for messages
 			// that are already in the cache, i.e. have been (recently) used.
-			if rm != nil {
-				as.setCachedRetainedMsg(key, rm, true, copyBytesToCache)
-			}
+			as.setCachedRetainedMsg(key, rm, true, copyBytesToCache)
 			return
 		}
 	}
@@ -2269,7 +2274,9 @@ func (as *mqttAccountSessionManager) handleRetainedMsgDel(subject string, seq ui
 		as.sl = NewSublistWithCache()
 	}
 	if erm, ok := as.retmsgs[subject]; ok {
-		as.rmsCache.Delete(subject)
+		if as.rmsCache != nil {
+			as.rmsCache.Delete(subject)
+		}
 		if erm.sub != nil {
 			as.sl.Remove(erm.sub)
 			erm.sub = nil
@@ -2770,6 +2777,7 @@ func (as *mqttAccountSessionManager) loadRetainedMessages(subjects map[string]st
 			w.Warnf("failed to decode retained message for subject %q: %v", ss[i], err)
 			continue
 		}
+
 		// Add the loaded retained message to the cache, and to the results map.
 		key := ss[i][len(mqttRetainedMsgsStreamSubject):]
 		as.setCachedRetainedMsg(key, &rm, false, false)
@@ -2960,6 +2968,9 @@ func (as *mqttAccountSessionManager) transferRetainedToPerKeySubjectStream(log *
 }
 
 func (as *mqttAccountSessionManager) getCachedRetainedMsg(subject string) *mqttRetainedMsg {
+	if as.rmsCache == nil {
+		return nil
+	}
 	v, ok := as.rmsCache.Load(subject)
 	if !ok {
 		return nil
@@ -2973,6 +2984,9 @@ func (as *mqttAccountSessionManager) getCachedRetainedMsg(subject string) *mqttR
 }
 
 func (as *mqttAccountSessionManager) setCachedRetainedMsg(subject string, rm *mqttRetainedMsg, onlyReplace bool, copyBytesToCache bool) {
+	if as.rmsCache == nil || rm == nil {
+		return
+	}
 	rm.expiresFromCache = time.Now().Add(mqttRetainedCacheTTL)
 	if onlyReplace {
 		if _, ok := as.rmsCache.Load(subject); !ok {


Test patch:
diff --git a/.github/workflows/MQTT_test.yaml b/.github/workflows/MQTT_test.yaml
index fe73d6e8..bfbd9210 100644
--- a/.github/workflows/MQTT_test.yaml
+++ b/.github/workflows/MQTT_test.yaml
@@ -1,15 +1,15 @@
-name: MQTT Compliance 
+name: MQTTEx
 on: [push, pull_request]
 
+permissions:
+  pull-requests: write # to comment on PRs
+  contents: write # to comment on commits (to upload artifacts?)
+
 jobs:
   test:
-    strategy:
-      matrix:
-        go: ["1.21"]
     env:
       GOPATH: /home/runner/work/nats-server
       GO111MODULE: "on"
-
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
@@ -20,18 +20,48 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v5
         with:
-          go-version: ${{matrix.go}}
+          go-version-file: src/github.com/nats-io/nats-server/go.mod
+          cache-dependency-path: src/github.com/nats-io/nats-server/go.sum
 
-      - name: Install deps
-        shell: bash --noprofile --norc -x -eo pipefail {0}
+      - name: Set up testing tools and environment
+        shell: bash --noprofile --norc -eo pipefail {0}
+        id: setup
         run: |
           wget https://github.com/hivemq/mqtt-cli/releases/download/v4.20.0/mqtt-cli-4.20.0.deb
           sudo apt install ./mqtt-cli-4.20.0.deb
+          go install github.com/ConnectEverything/mqtt-test@latest
+
+      # - name: Download benchmark result for ${{ github.base_ref || github.ref_name }}
+      #   uses: dawidd6/action-download-artifact@v2
+      #   continue-on-error: true
+      #   with:
+      #     path: src/github.com/nats-io/nats-server/bench
+      #     name: bench-output-${{ runner.os }}
+      #     branch: ${{ github.base_ref || github.ref }}
 
-      - name: Run tests
-        shell: bash --noprofile --norc -x -eo pipefail {0}
+      - name: Run tests and benchmarks
+        shell: bash --noprofile --norc -eo pipefail {0}
         run: |
-          set -e
-          cd src/github.com/nats-io/nats-server/server
-          go test -v -vet=off --run=TestMQTTCLICompliance
-          set +e
+          cd src/github.com/nats-io/nats-server
+          go test -v --run='MQTTEx' ./server
+          # go test --run='-' --count=10 --bench 'MQTT_' ./server | tee output.txt
+          # go test --run='-' --count=10 --bench 'MQTTEx' --timeout=20m --benchtime=100x ./server | tee -a output.txt
+          go test --run='-' --count=3 --bench 'MQTTEx' --benchtime=100x ./server
+
+      # - name: Compare benchmarks
+      #   uses: benchmark-action/github-action-benchmark@v1
+      #   with:
+      #     tool: go
+      #     output-file-path: src/github.com/nats-io/nats-server/output.txt
+      #     github-token: ${{ secrets.GITHUB_TOKEN }}
+      #     alert-threshold: 140%
+      #     comment-on-alert: true
+      #     # fail-on-alert: true
+      #     external-data-json-path: src/github.com/nats-io/nats-server/bench/benchmark-data.json
+
+      # - name: Store benchmark result for ${{ github.ref_name }}
+      #   if: always()
+      #   uses: actions/upload-artifact@v3
+      #   with:
+      #     path: src/github.com/nats-io/nats-server/bench
+      #     name: bench-output-${{ runner.os }}
diff --git a/server/mqtt_ex_test.go b/server/mqtt_ex_test.go
new file mode 100644
index 00000000..44025bef
--- /dev/null
+++ b/server/mqtt_ex_test.go
@@ -0,0 +1,424 @@
+// Copyright 2024 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !skip_mqtt_tests
+// +build !skip_mqtt_tests
+
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func TestMQTTExCompliance(t *testing.T) {
+	mqttPath := os.Getenv("MQTT_CLI")
+	if mqttPath == "" {
+		if p, err := exec.LookPath("mqtt"); err == nil {
+			mqttPath = p
+		}
+	}
+	if mqttPath == "" {
+		t.Skip(`"mqtt" command is not found in $PATH nor $MQTT_CLI. See https://hivemq.github.io/mqtt-cli/docs/installation/#debian-package for installation instructions`)
+	}
+
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		server_name: mqtt
+		jetstream {
+			store_dir = %q
+		}
+		mqtt {
+			listen: 127.0.0.1:-1
+		}
+	`, t.TempDir())))
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	cmd := exec.Command(mqttPath, "test", "-V", "3", "-p", strconv.Itoa(o.MQTT.Port))
+
+	output, err := cmd.CombinedOutput()
+	t.Log(string(output))
+	if err != nil {
+		if exitError, ok := err.(*exec.ExitError); ok {
+			t.Fatalf("mqtt cli exited with error: %v", exitError)
+		}
+	}
+}
+
+const (
+	KB = 1024
+)
+
+type mqttBenchMatrix struct {
+	QOS         []int
+	MessageSize []int
+	Topics      []int
+	Publishers  []int
+	Subscribers []int
+}
+
+type mqttBenchContext struct {
+	QOS         int
+	MessageSize int
+	Topics      int
+	Publishers  int
+	Subscribers int
+
+	Host string
+	Port int
+
+	// full path to mqtt-test command
+	testCmdPath string
+}
+
+var mqttBenchDefaultMatrix = mqttBenchMatrix{
+	QOS:         []int{0, 1, 2},
+	MessageSize: []int{100, 1 * KB, 10 * KB},
+	Topics:      []int{100},
+	Publishers:  []int{1},
+	Subscribers: []int{1},
+}
+
+type MQTTBenchmarkResult struct {
+	Ops   int                      `json:"ops"`
+	NS    map[string]time.Duration `json:"ns"`
+	Bytes int64                    `json:"bytes"`
+}
+
+func BenchmarkMQTTEx(b *testing.B) {
+	bc := mqttNewBenchEx(b)
+	b.Run("Server", func(b *testing.B) {
+		b.Cleanup(bc.startServer(b, false))
+		bc.runAll(b)
+	})
+
+	b.Run("Cluster", func(b *testing.B) {
+		b.Cleanup(bc.startCluster(b, false))
+		bc.runAll(b)
+	})
+
+	b.Run("Server-no-RMSCache", func(b *testing.B) {
+		b.Cleanup(bc.startServer(b, true))
+
+		bc.benchmarkSubRet(b)
+	})
+
+	b.Run("Cluster-no-RMSCache", func(b *testing.B) {
+		b.Cleanup(bc.startCluster(b, true))
+
+		bc.benchmarkSubRet(b)
+	})
+}
+
+func (bc mqttBenchContext) runAll(b *testing.B) {
+	bc.benchmarkPub(b)
+	bc.benchmarkPubRetained(b)
+	bc.benchmarkPubSub(b)
+	bc.benchmarkSubRet(b)
+}
+
+// makes a copy of bc
+func (bc mqttBenchContext) benchmarkPub(b *testing.B) {
+	m := mqttBenchDefaultMatrix.
+		NoSubscribers().
+		NoTopics()
+
+	b.Run("PUB", func(b *testing.B) {
+		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
+			bc.runCommand(b, "pub",
+				"--qos", strconv.Itoa(bc.QOS),
+				"--n", strconv.Itoa(b.N),
+				"--size", strconv.Itoa(bc.MessageSize),
+				"--num-publishers", strconv.Itoa(bc.Publishers),
+			)
+		})
+	})
+}
+
+// makes a copy of bc
+func (bc mqttBenchContext) benchmarkPubRetained(b *testing.B) {
+	// This bench is meaningless for QOS0 since the client considers the message
+	// sent as soon as it's written out. It is also useless for QOS2 since the
+	// flow takes a lot longer, and the difference of publishing as retained or
+	// not is lost in the noise.
+	m := mqttBenchDefaultMatrix.
+		NoSubscribers().
+		NoTopics().
+		QOS1Only()
+
+	b.Run("PUBRET", func(b *testing.B) {
+		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
+			bc.runCommand(b, "pub", "--retain",
+				"--qos", strconv.Itoa(bc.QOS),
+				"--n", strconv.Itoa(b.N),
+				"--size", strconv.Itoa(bc.MessageSize),
+				"--num-publishers", strconv.Itoa(bc.Publishers),
+			)
+		})
+	})
+}
+
+// makes a copy of bc
+func (bc mqttBenchContext) benchmarkPubSub(b *testing.B) {
+	// This test uses a single built-in topic, and a built-in publisher, so no
+	// reason to run it for topics and publishers.
+	m := mqttBenchDefaultMatrix.
+		NoTopics().
+		NoPublishers()
+
+	b.Run("PUBSUB", func(b *testing.B) {
+		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
+			bc.runCommand(b, "pubsub",
+				"--qos", strconv.Itoa(bc.QOS),
+				"--n", strconv.Itoa(b.N),
+				"--size", strconv.Itoa(bc.MessageSize),
+				"--num-subscribers", strconv.Itoa(bc.Subscribers),
+			)
+		})
+	})
+}
+
+// makes a copy of bc
+func (bc mqttBenchContext) benchmarkSubRet(b *testing.B) {
+	// This test uses a a built-in publisher, and it makes most sense to measure
+	// the retained message delivery "overhead" on a QoS0 subscription; without
+	// the extra time involved in actually subscribing.
+	m := mqttBenchDefaultMatrix.
+		NoPublishers().
+		QOS0Only()
+
+	b.Run("SUBRET", func(b *testing.B) {
+		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
+			bc.runCommand(b, "subret",
+				"--qos", strconv.Itoa(bc.QOS),
+				"--n", strconv.Itoa(b.N), // number of subscribe requests
+				"--num-subscribers", strconv.Itoa(bc.Subscribers),
+				"--num-topics", strconv.Itoa(bc.Topics),
+				"--size", strconv.Itoa(bc.MessageSize),
+			)
+		})
+	})
+}
+
+func mqttBenchLookupCommand(b *testing.B, name string) string {
+	b.Helper()
+	cmd, err := exec.LookPath(name)
+	if err != nil {
+		b.Skipf("%q command is not found in $PATH. Please `go install github.com/nats-io/meta-nats/apps/go/mqtt/...@latest` and try again.", name)
+	}
+	return cmd
+}
+
+func (bc mqttBenchContext) runCommand(b *testing.B, name string, extraArgs ...string) {
+	b.Helper()
+
+	args := append([]string{
+		name,
+		"-q",
+		"--servers", fmt.Sprintf("%s:%d", bc.Host, bc.Port),
+	}, extraArgs...)
+
+	cmd := exec.Command(bc.testCmdPath, args...)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		b.Fatalf("Error executing %q: %v", cmd.String(), err)
+	}
+	defer stdout.Close()
+	errbuf := bytes.Buffer{}
+	cmd.Stderr = &errbuf
+	if err = cmd.Start(); err != nil {
+		b.Fatalf("Error executing %q: %v", cmd.String(), err)
+	}
+	r := &MQTTBenchmarkResult{}
+	if err = json.NewDecoder(stdout).Decode(r); err != nil {
+		b.Fatalf("failed to decode output of %q: %v\n\n%s", cmd.String(), err, errbuf.String())
+	}
+	if err = cmd.Wait(); err != nil {
+		b.Fatalf("Error executing %q: %v\n\n%s", cmd.String(), err, errbuf.String())
+	}
+
+	r.report(b)
+}
+
+func (bc mqttBenchContext) initServer(b *testing.B) {
+	b.Helper()
+	bc.runCommand(b, "pubsub",
+		"--id", "__init__",
+		"--qos", "0",
+		"--n", "1",
+		"--size", "100",
+		"--num-subscribers", "1")
+}
+
+func (bc *mqttBenchContext) startServer(b *testing.B, disableRMSCache bool) func() {
+	b.Helper()
+	b.StopTimer()
+	prevDisableRMSCache := testDisableRMSCache
+	testDisableRMSCache = disableRMSCache
+	o := testMQTTDefaultOptions()
+	s := testMQTTRunServer(b, o)
+
+	o = s.getOpts()
+	bc.Host = o.MQTT.Host
+	bc.Port = o.MQTT.Port
+	bc.initServer(b)
+	return func() {
+		testMQTTShutdownServer(s)
+		testDisableRMSCache = prevDisableRMSCache
+	}
+}
+
+func (bc *mqttBenchContext) startCluster(b *testing.B, disableRMSCache bool) func() {
+	b.Helper()
+	b.StopTimer()
+	prevDisableRMSCache := testDisableRMSCache
+	testDisableRMSCache = disableRMSCache
+	conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			stream_replicas: 3
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+
+	cl := createJetStreamClusterWithTemplate(b, conf, "MQTT", 3)
+	o := cl.randomNonLeader().getOpts()
+	bc.Host = o.MQTT.Host
+	bc.Port = o.MQTT.Port
+	bc.initServer(b)
+	return func() {
+		cl.shutdown()
+		testDisableRMSCache = prevDisableRMSCache
+	}
+}
+
+func mqttBenchWrapForMatrixField(
+	vFieldPtr *int,
+	arr []int,
+	f func(b *testing.B, bc *mqttBenchContext),
+	namef func(int) string,
+) func(b *testing.B, bc *mqttBenchContext) {
+	if len(arr) == 0 {
+		return f
+	}
+	return func(b *testing.B, bc *mqttBenchContext) {
+		for _, value := range arr {
+			*vFieldPtr = value
+			b.Run(namef(value), func(b *testing.B) {
+				f(b, bc)
+			})
+		}
+	}
+}
+
+func (m mqttBenchMatrix) runMatrix(b *testing.B, bc mqttBenchContext, f func(*testing.B, *mqttBenchContext)) {
+	b.Helper()
+	f = mqttBenchWrapForMatrixField(&bc.MessageSize, m.MessageSize, f, func(size int) string {
+		return sizeKB(size)
+	})
+	f = mqttBenchWrapForMatrixField(&bc.Topics, m.Topics, f, func(n int) string {
+		return fmt.Sprintf("%dtopics", n)
+	})
+	f = mqttBenchWrapForMatrixField(&bc.Publishers, m.Publishers, f, func(n int) string {
+		return fmt.Sprintf("%dpubc", n)
+	})
+	f = mqttBenchWrapForMatrixField(&bc.Subscribers, m.Subscribers, f, func(n int) string {
+		return fmt.Sprintf("%dsubc", n)
+	})
+	f = mqttBenchWrapForMatrixField(&bc.QOS, m.QOS, f, func(qos int) string {
+		return fmt.Sprintf("QOS%d", qos)
+	})
+	b.ResetTimer()
+	b.StartTimer()
+	f(b, &bc)
+}
+
+func (m mqttBenchMatrix) NoSubscribers() mqttBenchMatrix {
+	m.Subscribers = nil
+	return m
+}
+
+func (m mqttBenchMatrix) NoTopics() mqttBenchMatrix {
+	m.Topics = nil
+	return m
+}
+
+func (m mqttBenchMatrix) NoPublishers() mqttBenchMatrix {
+	m.Publishers = nil
+	return m
+}
+
+func (m mqttBenchMatrix) QOS0Only() mqttBenchMatrix {
+	m.QOS = []int{0}
+	return m
+}
+
+func (m mqttBenchMatrix) QOS1Only() mqttBenchMatrix {
+	m.QOS = []int{1}
+	return m
+}
+
+func sizeKB(size int) string {
+	unit := "B"
+	N := size
+	if size >= KB {
+		unit = "KB"
+		N = (N + KB/2) / KB
+	}
+	return fmt.Sprintf("%d%s", N, unit)
+}
+
+func (r MQTTBenchmarkResult) report(b *testing.B) {
+	// Disable the default ns metric in favor of custom X_ms/op.
+	b.ReportMetric(0, "ns/op")
+
+	// Disable MB/s since the github benchmarking action misinterprets the sign
+	// of the result (treats it as less is better).
+	b.SetBytes(0)
+	// b.SetBytes(r.Bytes)
+
+	for unit, ns := range r.NS {
+		nsOp := float64(ns) / float64(r.Ops)
+		b.ReportMetric(nsOp/1000000, unit+"_ms/op")
+	}
+
+	// Diable ReportAllocs() since it confuses the github benchmarking action
+	// with the noise.
+	// b.ReportAllocs()
+}
+
+func mqttNewBenchEx(b *testing.B) *mqttBenchContext {
+	cmd := mqttBenchLookupCommand(b, "mqtt-test")
+	return &mqttBenchContext{
+		testCmdPath: cmd,
+	}
+}