Fix and client close tests

libbeat/publisher/pipeline/client.go

@@ -139,10 +139,14 @@ func (c *client) publish(e beat.Event) {
 }
 
 func (c *client) Close() error {
+	log := c.logger()
+
 	// first stop ack handling. ACK handler might block on wait (with timeout), waiting
 	// for pending events to be ACKed.
 	c.doClose()
+	log.Debug("client: wait for acker to finish")
 	c.acker.wait()
+	log.Debug("client: acker shut down")
 	return nil
 }
 

libbeat/publisher/pipeline/client_ack.go

@@ -51,9 +51,10 @@ func (p *Pipeline) makeACKer(
 		acker = bld.createEventACKer(canDrop, sema, cb)
 	default:
 		if waitClose <= 0 {
-			return bld.createPipelineACKer(canDrop, sema)
+			acker = bld.createPipelineACKer(canDrop, sema)
+		} else {
+			acker = bld.createCountACKer(canDrop, sema, func(_ int) {})
 		}
-		acker = bld.createCountACKer(canDrop, sema, func(_ int) {})
 	}
 
 	if waitClose <= 0 {

libbeat/publisher/pipeline/client_test.go

@@ -0,0 +1,115 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pipeline
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/outputs"
+	"github.com/elastic/beats/libbeat/publisher/queue"
+	"github.com/elastic/beats/libbeat/tests/resources"
+)
+
+func TestClient(t *testing.T) {
+	makePipeline := func(settings Settings, qu queue.Queue) *Pipeline {
+		p, err := New(beat.Info{},
+			Monitors{},
+			func(_ queue.Eventer) (queue.Queue, error) {
+				return qu, nil
+			},
+			outputs.Group{},
+			settings,
+		)
+		if err != nil {
+			panic(err)
+		}
+
+		return p
+	}
+
+	t.Run("client close", func(t *testing.T) {
+		// Note: no asserts. If closing fails we have a deadlock, because Publish
+		// would block forever
+
+		cases := map[string]struct {
+			context bool
+			close   func(client beat.Client, cancel func())
+		}{
+			"close unblocks client without context": {
+				context: false,
+				close: func(client beat.Client, _ func()) {
+					client.Close()
+				},
+			},
+			"close unblocks client with context": {
+				context: true,
+				close: func(client beat.Client, _ func()) {
+					client.Close()
+				},
+			},
+			"context cancel unblocks client": {
+				context: true,
+				close: func(client beat.Client, cancel func()) {
+					cancel()
+				},
+			},
+		}
+
+		if testing.Verbose() {
+			logp.TestingSetup()
+		}
+
+		for name, test := range cases {
+			t.Run(name, func(t *testing.T) {
+				routinesChecker := resources.NewGoroutinesChecker()
+				defer routinesChecker.Check(t)
+
+				pipeline := makePipeline(Settings{}, makeBlockingQueue())
+				defer pipeline.Close()
+
+				var ctx context.Context
+				var cancel func()
+				if test.context {
+					ctx, cancel = context.WithCancel(context.Background())
+				}
+
+				client, err := pipeline.ConnectWith(beat.ClientConfig{
+					CloseRef: ctx,
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer client.Close()
+
+				var wg sync.WaitGroup
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					client.Publish(beat.Event{})
+				}()
+
+				test.close(client, cancel)
+				wg.Wait()
+			})
+		}
+	})
+}

libbeat/publisher/pipeline/consumer.go

@@ -170,10 +170,12 @@ func (c *eventConsumer) loop(consumer queue.Consumer) {
 				consumer = nil
 				continue
 			}
+			if queueBatch != nil {
+				batch = newBatch(c.ctx, queueBatch, c.out.timeToLive)
+			}
 
-			batch = newBatch(c.ctx, queueBatch, c.out.timeToLive)
 			paused = c.paused()
-			if paused {
+			if paused || batch == nil {
 				out = nil
 			}
 		}

libbeat/publisher/pipeline/pipeline.go

@@ -278,7 +278,9 @@ func (p *Pipeline) Close() error {
 	}
 
 	p.observer.cleanup()
-	close(p.sigNewClient)
+	if p.sigNewClient != nil {
+		close(p.sigNewClient)
+	}
 
 	return nil
 }
@@ -366,7 +368,7 @@ func (p *Pipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error) {
 	if acker != nil {
 		producerCfg.ACK = acker.ackEvents
 	} else {
-		acker = nilACKer
+		acker = newCloseACKer(nilACKer, client.unlink)
 	}
 
 	client.acker = acker