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

@@ -616,10 +616,15 @@ type NatsAuthentication struct {
 	NatsTokenBasedAuthentication `yaml:"token,inline"`
 }
 
+type NatsConsumersConfiguration struct {
+	DeleteOnShutdown bool `yaml:"delete_on_shutdown" envDefault:"false"`
+}
+
 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"`
+	Consumers      NatsConsumersConfiguration `yaml:"consumers,omitempty"`
 }
 
 func (n NatsEventSource) GetID() string {

router/pkg/config/config.schema.json

@@ -2580,6 +2580,18 @@
                         }
                       }
                     ]
+                  },
+                  "consumers": {
+                    "type": "object",
+                    "description": "Configuration for durable JetStream consumers managed by this NATS provider.",
+                    "additionalProperties": false,
+                    "properties": {
+                      "delete_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,10 @@
         {
           "ID": "default",
           "URL": "nats://localhost:4222",
-          "Authentication": null
+          "Authentication": null,
+          "Consumers": {
+            "DeleteOnShutdown": false
+          }
         },
         {
           "ID": "my-nats",
@@ -667,6 +670,9 @@
               "Username": "admin"
             },
             "Token": null
+          },
+          "Consumers": {
+            "DeleteOnShutdown": false
           }
         }
       ],

router/pkg/pubsub/nats/adapter.go

@@ -32,6 +32,11 @@ type Adapter interface {
 // Ensure ProviderAdapter implements ProviderSubscriptionHooks
 var _ datasource.Adapter = (*ProviderAdapter)(nil)
 
+type consumerConfig struct {
+	deleteOnShutdown bool
+	trackedConsumers sync.Map // used as a set, key = consumer name, value = stream name
+}
+
 // ProviderAdapter implements the AdapterInterface for NATS pub/sub
 type ProviderAdapter struct {
 	ctx               context.Context
@@ -46,6 +51,7 @@ type ProviderAdapter struct {
 	opts              []nats.Option
 	flushTimeout      time.Duration
 	streamMetricStore metric.StreamMetricStore
+	consumerConfig    consumerConfig
 }
 
 // getInstanceIdentifier returns an identifier for the current instance.
@@ -116,6 +122,11 @@ 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)
 		}
 
+		// Track newly created durable consumer so we can later delete them, if it's necessary.
+		if p.consumerConfig.deleteOnShutdown {
+			p.consumerConfig.trackedConsumers.Store(durableConsumerName, subConf.StreamConfiguration.StreamName)
+		}
+
 		p.closeWg.Add(1)
 
 		go func() {
@@ -383,6 +394,34 @@ func (p *ProviderAdapter) Shutdown(ctx context.Context) error {
 
 	var shutdownErr error
 
+	// Delete durable consumers before closing the connection.
+	if p.consumerConfig.deleteOnShutdown && p.js != nil {
+		p.consumerConfig.trackedConsumers.Range(func(key, value any) bool {
+			consumerName, ok := key.(string)
+			if !ok {
+				// skip this odd element, should not happen in reality
+				return true
+			}
+
+			streamName, ok := value.(string)
+			if !ok {
+				// skip this odd element, should not happen in reality
+				return true
+			}
+
+			err := p.js.DeleteConsumer(ctx, streamName, consumerName)
+			if err != nil {
+				p.logger.Warn("failed to delete durable consumer on shutdown",
+					zap.String("stream", streamName),
+					zap.String("consumer", consumerName),
+					zap.Error(err),
+				)
+			}
+
+			return true
+		})
+	}
+
 	fErr := p.flush(ctx)
 	if fErr != nil {
 		shutdownErr = errors.Join(shutdownErr, fErr)
@@ -407,7 +446,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()
 	}
@@ -432,6 +471,9 @@ func NewAdapter(ctx context.Context, logger *zap.Logger, url string, opts []nats
 		opts:              opts,
 		flushTimeout:      10 * time.Second,
 		streamMetricStore: store,
+		consumerConfig: consumerConfig{
+			deleteOnShutdown: 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)
 	}