soaktest

README

Information for soak test for go client

The soak test contains 2 applications: soakclient.go for one application : Create Producer-1 to keep sending messages to the input-topic. Create Consumer-2 to receive messages from the output-topic.

soakclient_transaction.go for the other application: Create Consumer-1 to receive messages from the input-topic. Create transactional producer Producer-2 to sending the received messages to the output-topic. The initTransaction API registers a transactional.id when creating the producer. The producer sends the messages to the output-topic. Commit transaction after sending 100 messages each time or 1 second.

Instruction on configuration and running

Build soaktest API

$ cd soaktest
$ go mod init soaktest
$ go mod tidy
$ go install soaktest

Build and run soakclient.go

$ cd soaktest/soakclient
$ go build
$ ./soakclient -broker <broker> -inputTopic <..> -outTopic <..> -groupID <..> -inputTopicPartitionsNum <..> -outTopicPartitionsNum <..> -replicationFactor <..> -ccloudAPIKey <ccloudAPIKey> -ccloudAPISecret <ccloudAPISecret>

Build and run soakclient_transaction.go

$ cd soaktest/soakclient_transaction
$ go build
$ ./soakclient_transaction -broker <broker> -inputTopic <..> -outTopic <..> -outTopicPartitionsNum <..> -groupID <..> -transactionID <..> -ccloudAPIKey <ccloudAPIKey> -ccloudAPISecret <ccloudAPISecret>

The default values if doesn't input from the command lines

inputTopic: "inputTopic"
outTopic: "outTopic"
groupID : "groupID"
inputTopicPartitionsNum: 1
outTopicPartitionsNum: 1
replicationFactor: 1
transactionID: transactionID

Recommend instance type and size

At lease 4 CPUs and 8 GB volume size

Documentation

Index

• Constants
• Variables
• func ConvertUint64ToByteArray(num uint64) []byte
• func DatadogGauge(metricsName string, val float64, tags []string)
• func DatadogIncrement(metricsName string, incrval float64, tags []string)
• func GetRusage(metricsName string) error
• func GetRusageMetrics(metricsName string, wg *sync.WaitGroup, doneChan chan bool, ...)
• func HandleMessage(e *kafka.Message, hwmarks map[string]uint64) bool
• func HandleStatsEvent(e *kafka.Stats, topic string) error
• func InitLogFiles(filename string)
• func PrintConsumerStatus(consumer string)
• func PrintProducerStatus(producer string)
• func ProducerDeliveryCheck(e *kafka.Message)

Constants

const ConsumerError = "consumer.err"

ConsumerError combines the metrics name for consumer client-level errors

const FailedToProduceMsg = "failed.to.produce.message"

FailedToProduceMsg combines the metrics name if failed to produce messages

const ProducerError = "producer.err"

ProducerError combines the metrics name for producer client-level errors

Variables

var ConsumerErrCnt uint64

ConsumerErrCnt counts the number of consumer client-level errors

var ErrorLogger *log.Logger

ErrorLogger logs error level logs to the log file

var InfoLogger *log.Logger

InfoLogger logs info level logs to the log file

var ProduceMsgCnt uint64

ProduceMsgCnt is msg counter for producer

var ProducerErrCnt uint64

ProducerErrCnt counts the number of producer client-level errors

var WarningLogger *log.Logger

WarningLogger logs warning level logs to the log file

Functions

func ConvertUint64ToByteArray

func ConvertUint64ToByteArray(num uint64) []byte

ConvertUint64ToByteArray converts the unit64 type to []byte

func DatadogGauge

func DatadogGauge(metricsName string,
	val float64,
	tags []string)

DatadogGauge submits gauge type metrics

func DatadogIncrement

func DatadogIncrement(metricsName string, incrval float64, tags []string)

DatadogIncrement submits increment type metrics

func GetRusage

func GetRusage(metricsName string) error

GetRusage calculates RSS memory usage

func GetRusageMetrics

func GetRusageMetrics(metricsName string,
	wg *sync.WaitGroup,
	doneChan chan bool,
	errorChan chan error)

GetRusageMetrics is the entrance for resource calculation If caught a terminate signal, return, else call the GetRusage() function to calculate resource usage every 10 seconds

func HandleMessage

func HandleMessage(e *kafka.Message,
	hwmarks map[string]uint64) bool

HandleMessage handles received messages, monitors latency and verifies messages

func HandleStatsEvent

func HandleStatsEvent(e *kafka.Stats, topic string) error

HandleStatsEvent converts Stats events to map

func InitLogFiles

func InitLogFiles(filename string)

InitLogFiles initials the log folder and file, if the folder or file doesn't exist, create one, otherwise open it directly

func PrintConsumerStatus

func PrintConsumerStatus(consumer string)

PrintConsumerStatus prints the information for the consumer

func PrintProducerStatus

func PrintProducerStatus(producer string)

PrintProducerStatus prints the information for the producer

func ProducerDeliveryCheck

func ProducerDeliveryCheck(e *kafka.Message)

ProducerDeliveryCheck handles delivery report for producer