From 55b18d66ca82f2608433e0876d8b8158f2f13c32 Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Fri, 31 Jul 2026 11:47:38 -0400 Subject: [PATCH 1/3] fix(router): close Kafka consumer client when subscription ends ProviderAdapter.Subscribe creates a dedicated kgo.Client per subscription but never closes it. When a subscription ends the poller goroutine returns and the client is dropped without Close(); franz-go clients own background goroutines that keep the client reachable, so it is never garbage collected. Every ended subscription therefore permanently leaks a client together with its broker connections and buffered fetches, growing heap and goroutine count with the cumulative (not concurrent) subscription count until the process OOMs. A second issue makes it worse: topicPoller blocks in PollRecords on the adapter (application) context, not the subscription context, so a subscription cancelled while its topic is idle never unblocks the poller and neither the goroutine nor the client is ever reclaimed. Close the consumer client on every poller exit path via a sync.Once-guarded defer, and register a context.AfterFunc on the subscription context that closes the client on cancellation so an in-flight PollRecords returns IsClientClosed and the poller exits promptly. This also makes Shutdown reclaim consumer clients via the existing closeWg.Wait(). Co-Authored-By: Claude Opus 4.8 --- router/pkg/pubsub/kafka/adapter.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index 43dff3490..3f60ac049 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/router/pkg/pubsub/kafka/adapter.go @@ -166,6 +166,22 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri defer p.closeWg.Done() + // The consumer client owns background goroutines, broker connections and buffered + // fetches. It must be closed when the poller stops, otherwise every ended subscription + // leaks a full client for the lifetime of the process. Client.Close is not safe to call + // twice, so guard it with a sync.Once shared with the cancellation hook below. + var closeOnce sync.Once + closeClient := func() { closeOnce.Do(client.Close) } + defer closeClient() + + // topicPoller blocks in PollRecords on the adapter (application) context, not the + // subscription context. A subscription that is cancelled while its topic is idle would + // therefore never unblock the poller, so neither the goroutine nor the client would ever + // be reclaimed. Closing the client on subscription cancellation makes the in-flight + // PollRecords return IsClientClosed, so the poller exits promptly and the client is freed. + stopOnCancel := context.AfterFunc(ctx, closeClient) + defer stopOnCancel() + err := p.topicPoller(ctx, client, updater, PollerOpts{providerId: conf.ProviderID()}) if err != nil { if errors.Is(err, errClientClosed) || errors.Is(err, context.Canceled) { From 4c8226d152905ede0a4d426eec983e926b795bff Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Fri, 31 Jul 2026 14:35:13 -0400 Subject: [PATCH 2/3] refactor(router): use WaitGroup.Go for the Kafka subscription poller Address review feedback: replace the manual closeWg.Add(1) / go func() / defer closeWg.Done() with sync.WaitGroup.Go (Go 1.25). Behaviour is unchanged; the consumer-client cleanup logic is preserved verbatim. Co-Authored-By: Claude Opus 4.8 --- router/pkg/pubsub/kafka/adapter.go | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index 3f60ac049..85e3b02e9 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/router/pkg/pubsub/kafka/adapter.go @@ -160,11 +160,7 @@ 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. It must be closed when the poller stops, otherwise every ended subscription @@ -197,7 +193,7 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri } return } - }() + }) return nil } From fd6fdb22d4cb89a996dfb464d80066e20cba045c Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Thu, 6 Aug 2026 09:38:04 -0400 Subject: [PATCH 3/3] refactor(router): drive Kafka poller with a merged context Apply review feedback: instead of closing the client from a cancellation hook, derive the poller context from the subscription context and also cancel it when the adapter (application) context is cancelled, then pass that single context to topicPoller. topicPoller no longer references p.ctx (PollRecords and metric emission both use the passed context), so it returns immediately on a trigger close, router shutdown or hot reload. A single deferred client.Close() then reclaims the client, removing the need for the sync.Once guard. Co-Authored-By: Claude Opus 4.8 --- router/pkg/pubsub/kafka/adapter.go | 39 +++++++++++++----------------- 1 file changed, 17 insertions(+), 22 deletions(-) diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index 85e3b02e9..3ee51437f 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/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, @@ -161,24 +159,21 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri } p.closeWg.Go(func() { - // The consumer client owns background goroutines, broker connections and buffered - // fetches. It must be closed when the poller stops, otherwise every ended subscription - // leaks a full client for the lifetime of the process. Client.Close is not safe to call - // twice, so guard it with a sync.Once shared with the cancellation hook below. - var closeOnce sync.Once - closeClient := func() { closeOnce.Do(client.Close) } - defer closeClient() - - // topicPoller blocks in PollRecords on the adapter (application) context, not the - // subscription context. A subscription that is cancelled while its topic is idle would - // therefore never unblock the poller, so neither the goroutine nor the client would ever - // be reclaimed. Closing the client on subscription cancellation makes the in-flight - // PollRecords return IsClientClosed, so the poller exits promptly and the client is freed. - stopOnCancel := context.AfterFunc(ctx, closeClient) - defer stopOnCancel() - - err := p.topicPoller(ctx, client, updater, PollerOpts{providerId: conf.ProviderID()}) + // 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(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))