Fix race in publisher/pipeline retry loop (#13922)

* add WaitGroup to retry loop

* move workQueue close loop in controller

* tweak WG usage

* remove extra newlines

libbeat/publisher/pipeline/controller.go

@@ -84,6 +84,8 @@ func newOutputController(
 
 func (c *outputController) Close() error {
 	c.consumer.sigPause()
+	c.consumer.close()
+	c.retryer.close()
 
 	if c.out != nil {
 		for _, out := range c.out.outputs {
@@ -92,9 +94,6 @@ func (c *outputController) Close() error {
 		close(c.out.workQueue)
 	}
 
-	c.consumer.close()
-	c.retryer.close()
-
 	return nil
 }
 

libbeat/publisher/pipeline/retry.go

@@ -18,6 +18,8 @@
 package pipeline
 
 import (
+	"sync"
+
 	"github.com/elastic/beats/libbeat/logp"
 )
 
@@ -36,9 +38,10 @@ type retryer struct {
 
 	consumer *eventConsumer
 
-	sig chan retryerSignal
-	out workQueue
-	in  retryQueue
+	sig        chan retryerSignal
+	out        workQueue
+	in         retryQueue
+	doneWaiter sync.WaitGroup
 }
 
 type retryQueue chan batchEvent
@@ -75,20 +78,24 @@ func newRetryer(
 	c *eventConsumer,
 ) *retryer {
 	r := &retryer{
-		logger:   log,
-		observer: observer,
-		done:     make(chan struct{}),
-		sig:      make(chan retryerSignal, 3),
-		in:       retryQueue(make(chan batchEvent, 3)),
-		out:      out,
-		consumer: c,
+		logger:     log,
+		observer:   observer,
+		done:       make(chan struct{}),
+		sig:        make(chan retryerSignal, 3),
+		in:         retryQueue(make(chan batchEvent, 3)),
+		out:        out,
+		consumer:   c,
+		doneWaiter: sync.WaitGroup{},
 	}
+	r.doneWaiter.Add(1)
 	go r.loop()
 	return r
 }
 
 func (r *retryer) close() {
 	close(r.done)
+	//Block until loop() is properly closed
+	r.doneWaiter.Wait()
 }
 
 func (r *retryer) sigOutputAdded() {
@@ -115,6 +122,7 @@ func (r *retryer) cancelled(b *Batch) {
 }
 
 func (r *retryer) loop() {
+	defer r.doneWaiter.Done()
 	var (
 		out             workQueue
 		consumerBlocked bool
@@ -131,7 +139,6 @@ func (r *retryer) loop() {
 		select {
 		case <-r.done:
 			return
-
 		case evt := <-r.in:
 			var (
 				countFailed  int