testgen-kafka-parallelism.py

#!/usr/bin/env python

from __future__ import print_function
import json
import sys

def add_test():
    driver = {
        'name': 'Kafka',
        'driverClass': 'io.openmessaging.benchmark.driver.kafka.KafkaBenchmarkDriver',
        'replicationFactor': 3,

        'commonConfig':
          "bootstrap.servers=127.0.0.1",

        'topicConfig':
          "min.insync.replicas=2\n"
          # Uncomment below flags to enable sync mode in Kafka
          #"flush.messages=1\n"
          #"flush.ms=0\n"
          ,

        'producerConfig':
          "acks=all\n"
          "linger.ms=1\n"
          "batch.size=131072\n"
          # This is for transaction
          # "enableTransaction=True\n"
          # "eventPerTransaction=100\n"
          # "enable.idempotence=true"
          ,

        'consumerConfig':
          "auto.offset.reset=earliest\n"
          "enable.auto.commit=false",
      }

    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,
        driver=driver,
        workload=workload,
        numWorkers=numWorkers,
        localWorker=localWorker,
        tarball=tarball,
        build=build,
        undeploy=True,
    )
    test_list.append(t)


test_list = []

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

# Message size 1024 B
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 includeTimestampInEvent in [True]:
                                            add_test()

                                                
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)