[FIX]: filter_subject and filter_subjects are exclusive

Author: iapain

README.md

@@ -187,7 +187,7 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `delivery_group` - (optional) When set Push consumers will only deliver messages to subscriptions with this group set
  * `durable_name` - The durable name of the Consumer
  * `filter_subject` - (optional) Only receive a subset of messages from the Stream based on the subject they entered the Stream on
- * `filter_subjects` - (optional) Only receive a subset tof messages from the Stream based on subjects they entered the Stream on. Only works with v2.10 or better.
+ * `filter_subjects` - (optional) Only receive a subset tof messages from the Stream based on subjects they entered the Stream on. This is exclusive to `filter_subject`. Only works with v2.10 or better.
  * `max_delivery` - (optional) Maximum deliveries to attempt for each message
  * `replay_policy` - (optional) The rate at which messages will be replayed from the stream
  * `sample_freq` - (optional) The percentage of acknowledgements that will be sampled for observability purposes

docs/resources/jetstream_consumer.md

@@ -33,6 +33,7 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `delivery_group` - (optional) When set Push consumers will only deliver messages to subscriptions with this group set
  * `durable_name` - The durable name of the Consumer
  * `filter_subject` - (optional) Only receive a subset of messages from the Stream based on the subject they entered the Stream on
+ * `filter_subjects` - (optional) Only receive a subset tof messages from the Stream based on subjects they entered the Stream on. This is exclusive to `filter_subject`. Only works with v2.10 or better.
  * `max_delivery` - (optional) Maximum deliveries to attempt for each message
  * `replay_policy` - (optional) The rate at which messages will be replayed from the stream
  * `sample_freq` - (optional) The percentage of acknowledgements that will be sampled for observability purposes

jetstream/resource_jetstream_consumer.go

@@ -136,10 +136,11 @@ func resourceConsumer() *schema.Resource {
 				Default:     "",
 				Optional:    true,
 				ForceNew:    false,
+				ConflictsWith: []string{"filter_subjects"},
 			},
 			"filter_subjects": {
 				Type:        schema.TypeList,
-				Description: "Only receive a subset of messages from the stream baseed on the subjects they entered the Streeam on. Only works with nats-server v2.10+",
+				Description: "Only receive a subset of messages from the stream baseed on the subjects they entered the Streeam on, exlusive to filter_subject and works with nats-server v2.10 or better",
 				Optional:    true,
 				ForceNew:    false,
 				Elem:        &schema.Schema{Type: schema.TypeString},
@@ -336,12 +337,18 @@ func consumerConfigFromResourceData(d *schema.ResourceData) (cfg api.ConsumerCon
 
 	fs, ok := d.GetOk("filter_subjects")
 	if ok {
-		ar := fs.([]any)
-		var subjects = make([]string, len(ar))
-		for i, v := range ar {
-			subjects[i] = v.(string)
+		ns := fs.([]any)
+		if len(ns) == 1 {
+			cfg.FilterSubject = ns[0].(string)
+			cfg.FilterSubjects = nil
+		} else if len(ns) > 1 {
+			var subjects = make([]string, len(ns))
+			for i, v := range ns {
+				subjects[i] = v.(string)
+			}
+			cfg.FilterSubjects = subjects
+			cfg.FilterSubject = ""
 		}
-		cfg.FilterSubjects = subjects
 	}
 
 	m, ok := d.GetOk("metadata")
@@ -415,6 +422,8 @@ func resourceConsumerUpdate(d *schema.ResourceData, m any) error {
 	opts := []jsm.ConsumerOption{
 		jsm.ConsumerDescription(cfg.Description),
 		jsm.ConsumerMetadata(cfg.Metadata),
+		jsm.FilterStreamBySubject(cfg.FilterSubject),
+		jsm.FilterStreamBySubject(cfg.FilterSubjects...),
 		jsm.AckWait(cfg.AckWait),
 		jsm.MaxDeliveryAttempts(cfg.MaxDeliver),
 		jsm.SamplePercent(freq),

jetstream/resource_jetstream_consumer_test.go

@@ -53,6 +53,25 @@ resource "jetstream_consumer" "TEST_C2" {
 }
 `
 
+const testConsumerConfig_singleSubject = `
+provider "jetstream" {
+  servers = "%s"
+}
+
+resource "jetstream_stream" "test" {
+  name     = "TEST"
+  subjects = ["TEST.*"]
+}
+
+resource "jetstream_consumer" "TEST_C3" {
+  stream_id       = jetstream_stream.test.id
+  durable_name    = "C3"
+  stream_sequence = 10
+  max_ack_pending = 20
+  filter_subject = "TEST.a"
+}
+`
+
 func TestResourceConsumer(t *testing.T) {
 	srv := createJSServer(t)
 	defer srv.Shutdown()
@@ -108,6 +127,16 @@ func TestResourceConsumer(t *testing.T) {
 					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C2", "inactive_threshold", "0"),
 				),
 			},
+			{
+				Config: fmt.Sprintf(testConsumerConfig_singleSubject, nc.ConnectedUrl()),
+				Check: resource.ComposeTestCheckFunc(
+					testStreamExist(t, mgr, "TEST"),
+					testConsumerExist(t, mgr, "TEST", "C3"),
+					testConsumerHasFilterSubjects(t, mgr, "TEST", "C3", []string{}),
+					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C3", "filter_subject", "TEST.a"),
+				),
+			},
+
 		},
 	})
 }