Add InactiveThreshold parsing

internal/controller/consumer_controller.go

@@ -253,9 +253,6 @@ func consumerSpecToConfig(spec *api.ConsumerSpec) (*jetstream.ConsumerConfig, er
 		MemoryStorage:      spec.MemStorage,
 		FilterSubjects:     spec.FilterSubjects,
 		Metadata:           spec.Metadata,
-
-		// Explicitly set not (yet) mapped fields
-		InactiveThreshold: 0,
 	}
 
 	// DeliverPolicy
@@ -319,6 +316,14 @@ func consumerSpecToConfig(spec *api.ConsumerSpec) (*jetstream.ConsumerConfig, er
 		config.MaxRequestExpires = d
 	}
 
+	if spec.InactiveThreshold != "" {
+		d, err := time.ParseDuration(spec.InactiveThreshold)
+		if err != nil {
+			return nil, fmt.Errorf("invalid inactive threshold: %w", err)
+		}
+		config.InactiveThreshold = d
+	}
+
 	return config, nil
 }
 

internal/controller/suite_test.go

@@ -36,7 +36,7 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
-	jetstreamnatsiov1beta2 "github.com/nats-io/nack/pkg/jetstream/apis/jetstream/v1beta2"
+	api "github.com/nats-io/nack/pkg/jetstream/apis/jetstream/v1beta2"
 )
 
 // These tests use Ginkgo (BDD-style Go testing framework). Refer to
@@ -80,7 +80,7 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(cfg).NotTo(BeNil())
 
-	err = jetstreamnatsiov1beta2.AddToScheme(scheme.Scheme)
+	err = api.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
 
 	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})