Add requeue + edgecase worker response tests for waitgroup stability in serial pipelining

Reuse waitgroups within listener to reduce allocations

Author: adamsnoah98

kafkajobs/config.go

@@ -318,7 +318,6 @@ func (c *config) InitDefault(l *zap.Logger) ([]kgo.Opt, error) {
 		default:
 			return nil, errors.Errorf("unknown pipelining strategy: %s", c.ConsumerOpts.PipeliningStrategy)
 		}
-
 	}
 
 	if c.Ping == nil {

kafkajobs/item.go

@@ -118,9 +118,7 @@ func (i *Item) Ack() error {
 		return errors.Str("failed to acknowledge the JOB, the pipeline is probably stopped")
 	}
 
-	if i.doneWg != nil {
-		i.doneWg.Done()
-	}
+	i.done()
 
 	select {
 	case i.commitsCh <- i.record:
@@ -139,9 +137,7 @@ func (i *Item) NackWithOptions(requeue bool, _ int) error {
 		return errors.Str("failed to NackWithOptions the JOB, the pipeline is probably stopped")
 	}
 
-	if i.doneWg != nil {
-		i.doneWg.Done()
-	}
+	i.done()
 
 	if requeue {
 		err := i.Requeue(nil, 0)
@@ -204,6 +200,13 @@ func (i *Item) Respond(_ []byte, _ string) error {
 	return nil
 }
 
+func (i *Item) done() {
+	if i.doneWg != nil {
+		i.doneWg.Done()
+		i.doneWg = nil
+	}
+}
+
 func fromJob(job jobs.Message) *Item {
 	return &Item{
 		Job:     job.Name(),

kafkajobs/listener.go

@@ -4,15 +4,15 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/propagation"
 	"sync"
 	"sync/atomic"
 
 	"github.com/roadrunner-server/api/v4/plugins/v3/jobs"
 	"github.com/roadrunner-server/events"
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
 	"go.uber.org/zap"
 )
 
@@ -27,6 +27,12 @@ func (d *Driver) listen() error {
 	ctx, d.kafkaCancelCtx = context.WithCancel(context.Background())
 	d.mu.Unlock()
 
+	// only for serial pipelining
+	pipelineWg := &sync.WaitGroup{}
+	partitionWgPool := &sync.Pool{New: func() interface{} {
+		return &sync.WaitGroup{}
+	}}
+
 	defer func() {
 		d.log.Debug("kafka listener stopped")
 	}()
@@ -140,20 +146,20 @@ func (d *Driver) listen() error {
 
 		switch d.cfg.ConsumerOpts.PipeliningStrategy {
 		case SerialPipelining:
-			fetchWg := &sync.WaitGroup{}
 			fetches.EachPartition(func(partition kgo.FetchTopicPartition) {
-				itemWg := &sync.WaitGroup{}
+				itemWg := partitionWgPool.Get().(*sync.WaitGroup)
 
-				fetchWg.Go(func() {
+				pipelineWg.Go(func() {
 					partition.EachRecord(func(r *kgo.Record) {
 						itemWg.Add(1)
 						item := fromConsumer(r, d.requeueCh, d.recordsCh, itemWg, &d.stopped)
 						d.insertTracedItem(item)
 						itemWg.Wait()
+						partitionWgPool.Put(itemWg)
 					})
 				})
 			})
-			fetchWg.Wait()
+			pipelineWg.Wait()
 		case FanOutPipelining:
 			fetches.EachRecord(func(r *kgo.Record) {
 				item := fromConsumer(r, d.requeueCh, d.recordsCh, nil, &d.stopped)

tests/configs/.rr-kafka-serial-consumption-repeated-acknowledgement.yaml

@@ -0,0 +1,51 @@
+version: '3'
+
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+server:
+  command: "php php_test_files/jobs/jobs_repeat_acknowledgements.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+logs:
+  level: debug
+  encoding: console
+  mode: development
+
+kafka:
+  brokers: ["127.0.0.1:9092"]
+
+jobs:
+  num_pollers: 2
+  pipeline_size: 100000
+  timeout: 100
+  pool:
+    num_workers: 2
+    allocate_timeout: 60s
+    destroy_timeout: 60s
+
+  pipelines:
+    test:
+      driver: kafka
+      config:
+        priority: 1
+        auto_create_topics_enable: true
+
+        group_options:
+          group_id: bar
+          block_rebalance_on_poll: true
+
+        producer_options:
+          max_message_bytes: 1000
+          required_acks: LeaderAck
+          compression_codec: snappy
+          disable_idempotent: true
+
+        consumer_options:
+          topics: [ "serial-repeats-1" ]
+          pipelining_strategy: Serial
+          consumer_offset:
+            type: AtStart
+  consume: [ "test" ]
+  

tests/configs/.rr-kafka-serial-consumption-requeue.yaml

@@ -0,0 +1,52 @@
+version: '3'
+
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+server:
+  command: "php php_test_files/jobs/jobs_err.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+logs:
+  level: debug
+  encoding: console
+  mode: development
+
+kafka:
+  brokers: ["127.0.0.1:9092"]
+
+jobs:
+  num_pollers: 2
+  pipeline_size: 100000
+  timeout: 100
+  pool:
+    num_workers: 2
+    allocate_timeout: 60s
+    destroy_timeout: 60s
+
+  pipelines:
+    test:
+      driver: kafka
+      config:
+        priority: 1
+        auto_create_topics_enable: true
+
+        group_options:
+          group_id: bar
+          block_rebalance_on_poll: true
+
+        producer_options:
+          max_message_bytes: 1000
+          required_acks: LeaderAck
+          compression_codec: snappy
+          disable_idempotent: true
+
+        consumer_options:
+          topics: [ "serial-requeue-1" ]
+          pipelining_strategy: Serial
+          consumer_offset:
+            type: AtStart
+
+  consume: [ "test" ]
+