fix(router): close Kafka consumer client when subscription ends

router/pkg/pubsub/kafka/adapter.go

@@ -55,14 +55,12 @@ type PollerOpts struct {
 func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, updater datasource.SubscriptionEventUpdater, pollerOpts PollerOpts) error {
 	for {
 		select {
-		case <-p.ctx.Done(): // Close the poller if the application context was canceled
-			return p.ctx.Err()
-		case <-ctx.Done(): // Close the poller if the subscription context was canceled
+		case <-ctx.Done(): // Close the poller if the context was canceled (subscription ended, or router shutdown/hot reload)
 			return ctx.Err()
 
 		default:
 			// Try to fetch max records from any subscribed topics
-			fetches := client.PollRecords(p.ctx, 10_000)
+			fetches := client.PollRecords(ctx, 10_000)
 			if fetches.IsClientClosed() {
 				return errClientClosed
 			}
@@ -104,7 +102,7 @@ func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, u
 					headers[header.Key] = header.Value
 				}
 
-				p.streamMetricStore.Consume(p.ctx, metric.StreamsEvent{
+				p.streamMetricStore.Consume(ctx, metric.StreamsEvent{
 					ProviderId:          pollerOpts.providerId,
 					StreamOperationName: kafkaReceive,
 					ProviderType:        metric.ProviderTypeKafka,
@@ -160,13 +158,22 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri
 		return err
 	}
 
-	p.closeWg.Add(1)
-
-	go func() {
-
-		defer p.closeWg.Done()
+	p.closeWg.Go(func() {
+		// The consumer client owns background goroutines, broker connections and buffered
+		// fetches, so it must be closed when the poller stops, otherwise every ended
+		// subscription leaks a full client for the lifetime of the process.
+		defer client.Close()
+
+		// Drive the poller with a context that is cancelled when EITHER the subscription
+		// context (ctx) or the adapter/application context (p.ctx) is cancelled. This makes
+		// topicPoller return immediately on a trigger close, router shutdown or hot reload,
+		// at which point the deferred Close above reclaims the client.
+		pollerCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+		stop := context.AfterFunc(p.ctx, cancel)
+		defer stop()
 
-		err := p.topicPoller(ctx, client, updater, PollerOpts{providerId: conf.ProviderID()})
+		err := p.topicPoller(pollerCtx, client, updater, PollerOpts{providerId: conf.ProviderID()})
 		if err != nil {
 			if errors.Is(err, errClientClosed) || errors.Is(err, context.Canceled) {
 				log.Debug("poller canceled", zap.Error(err))
@@ -181,7 +188,7 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri
 			}
 			return
 		}
-	}()
+	})
 
 	return nil
 }