aboutsummaryrefslogtreecommitdiff
path: root/test/kafka/loadtest/quick_performance_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'test/kafka/loadtest/quick_performance_test.go')
-rw-r--r--test/kafka/loadtest/quick_performance_test.go139
1 files changed, 139 insertions, 0 deletions
diff --git a/test/kafka/loadtest/quick_performance_test.go b/test/kafka/loadtest/quick_performance_test.go
new file mode 100644
index 000000000..299a7d948
--- /dev/null
+++ b/test/kafka/loadtest/quick_performance_test.go
@@ -0,0 +1,139 @@
+package integration
+
+import (
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// TestQuickPerformance_10K tests the fixed broker with 10K records
+func TestQuickPerformance_10K(t *testing.T) {
+ const (
+ totalRecords = 10000 // 10K records for quick test
+ numPartitions = int32(4)
+ numProducers = 4
+ brokerAddr = "localhost:17777"
+ )
+
+ // Create direct broker client
+ client, err := NewDirectBrokerClient(brokerAddr)
+ if err != nil {
+ t.Fatalf("Failed to create direct broker client: %v", err)
+ }
+ defer client.Close()
+
+ topicName := fmt.Sprintf("quick-test-%d", time.Now().Unix())
+
+ // Create topic
+ glog.Infof("Creating topic %s with %d partitions", topicName, numPartitions)
+ err = client.ConfigureTopic(topicName, numPartitions)
+ if err != nil {
+ t.Fatalf("Failed to configure topic: %v", err)
+ }
+
+ // Performance tracking
+ var totalProduced int64
+ var totalErrors int64
+ startTime := time.Now()
+
+ // Producer function
+ producer := func(producerID int, recordsPerProducer int) error {
+ for i := 0; i < recordsPerProducer; i++ {
+ recordID := producerID*recordsPerProducer + i
+
+ // Generate test record
+ testRecord := GenerateMockTestRecord(recordID)
+ key, value := SerializeMockTestRecord(testRecord)
+
+ partition := int32(testRecord.UserID % int64(numPartitions))
+
+ // Produce the record (now async!)
+ err := client.PublishRecord(topicName, partition, key, value)
+ if err != nil {
+ atomic.AddInt64(&totalErrors, 1)
+ if atomic.LoadInt64(&totalErrors) < 5 {
+ glog.Errorf("Producer %d failed to produce record %d: %v", producerID, recordID, err)
+ }
+ continue
+ }
+
+ atomic.AddInt64(&totalProduced, 1)
+
+ // Log progress
+ if (i+1)%1000 == 0 {
+ elapsed := time.Since(startTime)
+ rate := float64(atomic.LoadInt64(&totalProduced)) / elapsed.Seconds()
+ glog.Infof("Producer %d: %d records, current rate: %.0f records/sec",
+ producerID, i+1, rate)
+ }
+ }
+ return nil
+ }
+
+ // Start concurrent producers
+ glog.Infof("Starting %d producers for %d records total", numProducers, totalRecords)
+
+ var wg sync.WaitGroup
+ recordsPerProducer := totalRecords / numProducers
+
+ for i := 0; i < numProducers; i++ {
+ wg.Add(1)
+ go func(producerID int) {
+ defer wg.Done()
+ if err := producer(producerID, recordsPerProducer); err != nil {
+ glog.Errorf("Producer %d failed: %v", producerID, err)
+ }
+ }(i)
+ }
+
+ // Wait for completion
+ wg.Wait()
+
+ produceTime := time.Since(startTime)
+ finalProduced := atomic.LoadInt64(&totalProduced)
+ finalErrors := atomic.LoadInt64(&totalErrors)
+
+ // Performance results
+ throughputPerSec := float64(finalProduced) / produceTime.Seconds()
+ dataVolumeMB := float64(finalProduced) * 300 / (1024 * 1024) // ~300 bytes per record
+ throughputMBPerSec := dataVolumeMB / produceTime.Seconds()
+
+ glog.Infof("\n"+
+ "QUICK PERFORMANCE TEST RESULTS\n"+
+ "=====================================\n"+
+ "Records produced: %d / %d\n"+
+ "Production time: %v\n"+
+ "Throughput: %.0f records/sec\n"+
+ "Data volume: %.1f MB\n"+
+ "Bandwidth: %.1f MB/sec\n"+
+ "Errors: %d (%.2f%%)\n"+
+ "Success rate: %.1f%%\n",
+ finalProduced, totalRecords,
+ produceTime,
+ throughputPerSec,
+ dataVolumeMB,
+ throughputMBPerSec,
+ finalErrors,
+ float64(finalErrors)/float64(totalRecords)*100,
+ float64(finalProduced)/float64(totalRecords)*100,
+ )
+
+ // Assertions
+ if finalProduced < int64(totalRecords*0.90) { // Allow 10% tolerance
+ t.Errorf("Too few records produced: %d < %d (90%% of target)", finalProduced, int64(float64(totalRecords)*0.90))
+ }
+
+ if throughputPerSec < 100 { // Should be much higher than 1 record/sec now!
+ t.Errorf("Throughput too low: %.0f records/sec (expected > 100)", throughputPerSec)
+ }
+
+ if finalErrors > int64(totalRecords*0.10) { // Error rate should be < 10%
+ t.Errorf("Too many errors: %d > %d (10%% of target)", finalErrors, int64(float64(totalRecords)*0.10))
+ }
+
+ glog.Infof("Performance test passed! Ready for million-record test.")
+}