feat: option to remove nats consumers on shutdown

demo/pkg/subgraphs/subgraphs.go

@@ -212,7 +212,7 @@ func New(ctx context.Context, config *Config) (*Subgraphs, error) {
 
 	natsPubSubByProviderID := map[string]natsPubsub.Adapter{}
 
-	defaultAdapter, err := natsPubsub.NewAdapter(ctx, zap.NewNop(), url, []nats.Option{}, "hostname", "test", datasource.ProviderOpts{
+	defaultAdapter, err := natsPubsub.NewAdapter(ctx, zap.NewNop(), url, []nats.Option{}, "hostname", "test", false, datasource.ProviderOpts{
 		StreamMetricStore: rmetric.NewNoopStreamMetricStore(),
 	})
 	if err != nil {
@@ -223,7 +223,7 @@ func New(ctx context.Context, config *Config) (*Subgraphs, error) {
 	}
 	natsPubSubByProviderID["default"] = defaultAdapter
 
-	myNatsAdapter, err := natsPubsub.NewAdapter(ctx, zap.NewNop(), url, []nats.Option{}, "hostname", "test", datasource.ProviderOpts{
+	myNatsAdapter, err := natsPubsub.NewAdapter(ctx, zap.NewNop(), url, []nats.Option{}, "hostname", "test", false, datasource.ProviderOpts{
 		StreamMetricStore: rmetric.NewNoopStreamMetricStore(),
 	})
 	if err != nil {

router-tests/testenv/testenv.go

@@ -2911,7 +2911,7 @@ func subgraphOptions(ctx context.Context, t testing.TB, logger *zap.Logger, nats
 	}
 	natsPubSubByProviderID := make(map[string]pubsubNats.Adapter, len(DemoNatsProviders))
 	for _, sourceName := range DemoNatsProviders {
-		adapter, err := pubsubNats.NewAdapter(ctx, logger, natsData.Params[0].Url, natsData.Params[0].Opts, "hostname", "listenaddr", datasource.ProviderOpts{
+		adapter, err := pubsubNats.NewAdapter(ctx, logger, natsData.Params[0].Url, natsData.Params[0].Opts, "hostname", "listenaddr", false, datasource.ProviderOpts{
 			StreamMetricStore: rmetric.NewNoopStreamMetricStore(),
 		})
 		require.NoError(t, err)

router/pkg/config/config.go

@@ -617,9 +617,10 @@ type NatsAuthentication struct {
 }
 
 type NatsEventSource struct {
-	ID             string              `yaml:"id,omitempty"`
-	URL            string              `yaml:"url,omitempty"`
-	Authentication *NatsAuthentication `yaml:"authentication,omitempty"`
+	ID                               string              `yaml:"id,omitempty"`
+	URL                              string              `yaml:"url,omitempty"`
+	Authentication                   *NatsAuthentication `yaml:"authentication,omitempty"`
+	DeleteDurableConsumersOnShutdown bool                `yaml:"delete_durable_consumers_on_shutdown" envDefault:"false"`
 }
 
 func (n NatsEventSource) GetID() string {

router/pkg/config/config.schema.json

@@ -2580,6 +2580,11 @@
                         }
                       }
                     ]
+                  },
+                  "delete_durable_consumers_on_shutdown": {
+                    "type": "boolean",
+                    "description": "When enabled, all durable JetStream consumers created by this provider are deleted when the router shuts down. This prevents zombie consumers accumulating on the NATS server across Kubernetes rolling deployments where each pod has a unique hostname baked into the consumer name. Defaults to false.",
+                    "default": false
                   }
                 }
               }

router/pkg/config/testdata/config_full.json

@@ -656,7 +656,8 @@
         {
           "ID": "default",
           "URL": "nats://localhost:4222",
-          "Authentication": null
+          "Authentication": null,
+          "DeleteDurableConsumersOnShutdown": false
         },
         {
           "ID": "my-nats",
@@ -667,7 +668,8 @@
               "Username": "admin"
             },
             "Token": null
-          }
+          },
+          "DeleteDurableConsumersOnShutdown": false
         }
       ],
       "Kafka": [

router/pkg/pubsub/nats/adapter.go

@@ -32,6 +32,11 @@ type Adapter interface {
 // Ensure ProviderAdapter implements ProviderSubscriptionHooks
 var _ datasource.Adapter = (*ProviderAdapter)(nil)
 
+type trackedConsumer struct {
+	streamName   string
+	consumerName string
+}
+
 // ProviderAdapter implements the AdapterInterface for NATS pub/sub
 type ProviderAdapter struct {
 	ctx               context.Context
@@ -46,6 +51,10 @@ type ProviderAdapter struct {
 	opts              []nats.Option
 	flushTimeout      time.Duration
 	streamMetricStore metric.StreamMetricStore
+
+	deleteConsumersOnShutdown bool
+	trackedConsumersMu        sync.Mutex
+	trackedConsumers          []trackedConsumer
 }
 
 // getInstanceIdentifier returns an identifier for the current instance.
@@ -116,6 +125,15 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, cfg datasource.Subscrip
 			return datasource.NewError(fmt.Sprintf(`failed to create or update consumer for stream "%s"`, subConf.StreamConfiguration.StreamName), err)
 		}
 
+		if p.deleteConsumersOnShutdown {
+			p.trackedConsumersMu.Lock()
+			p.trackedConsumers = append(p.trackedConsumers, trackedConsumer{
+				streamName:   subConf.StreamConfiguration.StreamName,
+				consumerName: durableConsumerName,
+			})
+			p.trackedConsumersMu.Unlock()
+		}
+
 		p.closeWg.Add(1)
 
 		go func() {
@@ -383,6 +401,23 @@ func (p *ProviderAdapter) Shutdown(ctx context.Context) error {
 
 	var shutdownErr error
 
+	// Delete durable consumers before closing the connection — the JetStream
+	// API requires an open connection to reach the NATS server.
+	if p.deleteConsumersOnShutdown && p.js != nil {
+		p.trackedConsumersMu.Lock()
+		consumers := p.trackedConsumers
+		p.trackedConsumersMu.Unlock()
+		for _, tc := range consumers {
+			if err := p.js.DeleteConsumer(ctx, tc.streamName, tc.consumerName); err != nil {
+				p.logger.Warn("failed to delete durable consumer on shutdown",
+					zap.String("stream", tc.streamName),
+					zap.String("consumer", tc.consumerName),
+					zap.Error(err),
+				)
+			}
+		}
+	}
+
 	fErr := p.flush(ctx)
 	if fErr != nil {
 		shutdownErr = errors.Join(shutdownErr, fErr)
@@ -407,7 +442,7 @@ func (p *ProviderAdapter) Shutdown(ctx context.Context) error {
 	return nil
 }
 
-func NewAdapter(ctx context.Context, logger *zap.Logger, url string, opts []nats.Option, hostName string, routerListenAddr string, providerOpts datasource.ProviderOpts) (Adapter, error) {
+func NewAdapter(ctx context.Context, logger *zap.Logger, url string, opts []nats.Option, hostName string, routerListenAddr string, deleteConsumersOnShutdown bool, providerOpts datasource.ProviderOpts) (Adapter, error) {
 	if logger == nil {
 		logger = zap.NewNop()
 	}
@@ -422,16 +457,17 @@ func NewAdapter(ctx context.Context, logger *zap.Logger, url string, opts []nats
 	ctx, cancelFunc := context.WithCancel(ctx)
 
 	return &ProviderAdapter{
-		ctx:               ctx,
-		cancel:            cancelFunc,
-		logger:            logger.With(zap.String("pubsub", "nats")),
-		closeWg:           sync.WaitGroup{},
-		hostName:          hostName,
-		routerListenAddr:  routerListenAddr,
-		url:               url,
-		opts:              opts,
-		flushTimeout:      10 * time.Second,
-		streamMetricStore: store,
+		ctx:                       ctx,
+		cancel:                    cancelFunc,
+		logger:                    logger.With(zap.String("pubsub", "nats")),
+		closeWg:                   sync.WaitGroup{},
+		hostName:                  hostName,
+		routerListenAddr:          routerListenAddr,
+		url:                       url,
+		opts:                      opts,
+		flushTimeout:              10 * time.Second,
+		streamMetricStore:         store,
+		deleteConsumersOnShutdown: deleteConsumersOnShutdown,
 	}, nil
 }
 

router/pkg/pubsub/nats/provider_builder.go

@@ -123,7 +123,7 @@ func buildProvider(ctx context.Context, provider config.NatsEventSource, logger
 		return nil, fmt.Errorf("failed to build options for Nats provider with ID \"%s\": %w", provider.ID, err)
 	}
 
-	adapter, err := NewAdapter(ctx, logger, provider.URL, options, hostName, routerListenAddr, providerOpts)
+	adapter, err := NewAdapter(ctx, logger, provider.URL, options, hostName, routerListenAddr, provider.DeleteDurableConsumersOnShutdown, providerOpts)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create adapter for Nats provider with ID \"%s\": %w", provider.ID, err)
 	}