Cherry-pick #12063 to 6.8: Fix memory leak in Filebeat pipeline acker (…

…#12138)

* Fix memory leak in Filebeat pipeline acker (#12063)

* Fix memory leak in Filebeat pipeline acker

Before this change acker goroutine was kept forever as processed events
count was not correctly updated.

Filebeat sends an empty event to update file states, this event is not
published, but treated as dropped, without updating counters.

This change makes sures that `a.events` count gets updated for dropped
events also, so the acker gets closed after all ACKs happen.

(cherry picked from commit 9653105)

* Update CHANGELOG.next.asciidoc

CHANGELOG.next.asciidoc

@@ -44,6 +44,7 @@ https://github.com/elastic/beats/compare/v6.7.2...6.8[Check the HEAD diff]
 
 - Fix goroutine leak happening when harvesters are dynamically stopped. {pull}11263[11263]
 - Fix initialization of the TCP input logger. {pull}11605[11605]
+- Fix memory leak in Filebeat pipeline acker. {pull}12063[12063]
 
 *Heartbeat*
 

libbeat/publisher/pipeline/acker.go

@@ -139,19 +139,26 @@ func (a *gapCountACK) ackLoop() {
 		case <-a.done:
 			closing = true
 			a.done = nil
+			if a.events.Load() == 0 {
+				// stop worker, if all events accounted for have been ACKed.
+				// If new events are added after this acker won't handle them, which may
+				// result in duplicates
+				return
+			}
 
 		case <-a.pipeline.ackDone:
 			return
 
 		case n := <-acks:
 			empty := a.handleACK(n)
 			if empty && closing && a.events.Load() == 0 {
-				// stop worker, iff all events accounted for have been ACKed
+				// stop worker, if and only if all events accounted for have been ACKed
 				return
 			}
 
 		case <-drop:
 			// TODO: accumulate multiple drop events + flush count with timer
+			a.events.Sub(1)
 			a.fn(1, 0)
 		}
 	}