testgen-pulsar-parallelism.py

#!/usr/bin/env python

from __future__ import print_function
import json
import sys

test_list = []

localWorker = False
tarball = 'package/target/openmessaging-benchmark-0.0.1-SNAPSHOT-bin.tar.gz'
build = False
numWorkers = 10

for repeat in range(1):
    for testDurationMinutes in [4]:
        for messageSize in [1024]:
            for producerRateEventsPerSec in [25e4]:
                for topics in [1]:
                    for partitionsPerTopic in [1, 10, 100, 500, 1000, 5000]:
                        for producersPerTopic in [1, 10, 50, 100]:
                            if producersPerTopic == 1:
                                numWorkers = 2
                            else: numWorkers = 10
                            for consumerBacklogSizeGB in [0]:
                                for subscriptionsPerTopic in [1]:
                                    for consumerPerSubscription in [producersPerTopic]:
                                        for ackQuorum in [2]:   # 2 causes OOM in Bookie at max rate
                                            driver = {
                                                'name': 'Pulsar',
                                                'driverClass': 'io.openmessaging.benchmark.driver.pulsar.PulsarBenchmarkDriver',
                                                'enableTiering': True,
                                                'client': {
                                                           'ioThreads': 8,
                                                           'connectionsPerBroker': 8,
                                                           'clusterName': 'local',
                                                           'namespacePrefix': 'benchmark/ns',
                                                           'topicType': 'persistent',
                                                           'persistence': {'ensembleSize': 3,
                                                                           'writeQuorum': 3,
                                                                           'ackQuorum': ackQuorum,
                                                                           'deduplicationEnabled': True},
                                                           'tlsAllowInsecureConnection': False,
                                                           'tlsEnableHostnameVerification': False,
                                                           'tlsTrustCertsFilePath': None,
                                                           'authentication': {'plugin': None, 'data': None}},
                                                'producer': {'batchingEnabled': True,
                                                             'batchingMaxPublishDelayMs': 1,
                                                             'blockIfQueueFull': True,
                                                             'pendingQueueSize': 10000},
                                            }
                                            workload = {
                                                'messageSize': messageSize,
                                                'topics':  topics,
                                                'partitionsPerTopic': partitionsPerTopic,
                                                'subscriptionsPerTopic': subscriptionsPerTopic,
                                                'consumerPerSubscription': consumerPerSubscription,
                                                'producersPerTopic': producersPerTopic,
                                                'producerRate': producerRateEventsPerSec,
                                                'consumerBacklogSizeGB': consumerBacklogSizeGB,
                                                'testDurationMinutes': testDurationMinutes,
                                                'keyDistributor': 'RANDOM_NANO',
                                            }
                                            t = dict(
                                                test='openmessaging-benchmark',
                                                max_test_attempts=1,
                                                result_filename='data/results/json/%(test)s_%(test_uuid)s.json',
                                                driver=driver,
                                                workload=workload,
                                                numWorkers=numWorkers,
                                                localWorker=localWorker,
                                                tarball=tarball,
                                                build=build,
                                                undeploy=True,
                                            )
                                            test_list += [t]
                                            build = False


print(json.dumps(test_list, sort_keys=True, indent=4, ensure_ascii=False))
print('Number of tests generated: %d' % len(test_list), file=sys.stderr)