review

x-pack/filebeat/input/cloudfoundry/input_integration_test.go

@@ -21,7 +21,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
 	pubtest "github.com/elastic/beats/v7/libbeat/publisher/testing"
-	"github.com/elastic/beats/v7/libbeat/tests/resources"
 	cftest "github.com/elastic/beats/v7/x-pack/libbeat/common/cloudfoundry/test"
 )
 
@@ -38,8 +37,6 @@ func TestInput(t *testing.T) {
 }
 
 func testInput(t *testing.T, version string) {
-	defer resources.NewGoroutinesChecker().Check(t)
-
 	config := common.MustNewConfigFrom(cftest.GetConfigFromEnv(t))
 	config.SetString("version", -1, version)
 

x-pack/filebeat/input/cloudfoundry/v1.go

@@ -24,9 +24,10 @@ func configureV1(config cloudfoundry.Config) (*inputV1, error) {
 
 func (i *inputV1) Name() string { return "cloudfoundry-v1" }
 
-func (i *inputV1) Test(_ v2.TestContext) error {
-	// XXX: try to connect, but don't consume
-	return nil
+func (i *inputV1) Test(ctx v2.TestContext) error {
+	hub := cloudfoundry.NewHub(&i.config, "filebeat", ctx.Logger)
+	_, err := hub.Client()
+	return err
 }
 
 func (i *inputV1) Run(ctx v2.Context, publisher stateless.Publisher) error {
@@ -50,11 +51,13 @@ func (i *inputV1) Run(ctx v2.Context, publisher stateless.Publisher) error {
 		return errors.Wrapf(err, "initializing doppler consumer")
 	}
 
-	_, cancel := ctxtool.WithFunc(ctxtool.FromCanceller(ctx.Cancelation), func() {
-		consumer.Stop()
+	stopCtx, cancel := ctxtool.WithFunc(ctxtool.FromCanceller(ctx.Cancelation), func() {
+		// wait stops the consumer and waits for all internal go-routines to be stopped.
+		consumer.Wait()
 	})
 	defer cancel()
 
 	consumer.Run()
+	<-stopCtx.Done()
 	return nil
 }

x-pack/libbeat/common/cloudfoundry/rlplistener.go

@@ -66,8 +66,8 @@ func (c *RlpListener) Start(ctx context.Context) {
 	}
 	es := rlpClient.Stream(ctx, l)
 
+	c.wg.Add(1)
 	go func() {
-		c.wg.Add(1)
 		defer c.wg.Done()
 		for {
 			select {