fix: Kafka client resource leak

Sarama Kafka clients leak several resources (goroutines, metrics,
connections) if not closed properly. A client is created and managed
within the topic manager and therefore a topic manager will leak
resources if not properly closed.

Topic Managers are created in two places:

1. When calling `prepareTopics` while instantiating a new processor.
2. Within `processor.Run`.

The first of these is correctly closed, but the latter was not. Closing
the second instance should ensure underlying Sarama resources are
appropriate cleaned up when a processor stops.

Author: jalaziz

processor.go

@@ -288,6 +288,11 @@ func (g *Processor) Run(ctx context.Context) (rerr error) {
 	if err != nil {
 		return fmt.Errorf("Error creating topic manager for brokers [%s]: %v", strings.Join(g.brokers, ","), err)
 	}
+	defer func() {
+		if err := g.tmgr.Close(); err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("error closing topic manager: %w", err))
+		}
+	}()
 
 	// create kafka producer
 	g.log.Debugf("creating producer")

processor_test.go

@@ -220,7 +220,7 @@ func TestProcessor_Run(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)
 
@@ -263,7 +263,7 @@ func TestProcessor_Run(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)
 
@@ -300,7 +300,7 @@ func TestProcessor_Run(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)
 
@@ -343,7 +343,7 @@ func TestProcessor_Run(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)
 
@@ -394,7 +394,7 @@ func TestProcessor_Stop(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)
 
@@ -448,7 +448,7 @@ func TestProcessor_Stop(t *testing.T) {
 		ctrl, bm := createMockBuilder(t)
 		defer ctrl.Finish()
 
-		bm.tmgr.EXPECT().Close().Times(1)
+		bm.tmgr.EXPECT().Close().Times(2)
 		bm.tmgr.EXPECT().Partitions(gomock.Any()).Return([]int32{0}, nil).Times(1)
 		bm.producer.EXPECT().Close().Times(1)