Jetstream keeps producing messages even when not intended

[kharabela]

My producer code

func JetStreamAccountInit(data string) { 
	nc, err := nats.Connect(nats.DefaultURL, nats.ReconnectWait(10*time.Second))

	if err != nil {
		log.Fatal(err)
	}
	defer nc.Drain()

	ctx := context.Background()

	getStatusTxt := func(nc *nats.Conn) string {
		switch nc.Status() {
		case nats.CONNECTED:
			return "Connected"
		case nats.CLOSED:
			return "Closed"
		default:
			return "Other"
		}
	}
	log.Printf("The connection is %v\n", getStatusTxt(nc))

	// Use the JetStream context to produce and consumer messages that have been persisted.
	//js, err := nc.JetStream(nats.PublishAsyncMaxPending(10000)) //ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}

	// Create a stream if it does not exist

	_, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:        "Subjects",
		Description: "Messages for Account",
		Subjects: []string{
			"Subjects.AccountInfo",
		},
		//MaxBytes: 1024 * 1024 * 1024,
	})

	if err != nil {
		log.Fatal(err)
	}

	js.PublishAsync("Subjects.AccountInfo", []byte(data))
	select {
	case <-js.PublishAsyncComplete():
	case <-time.After(5 * time.Second):
		fmt.Println("Did not resolve in time")
	}

}

I didn't even was running my producer code.Just had my consumer code running to view messages.
I saw newly generated messages appearing in current time stamp
I don't understand why?

2024/07/21 12:00:54 Consumer Data =>  Page: Account_UI  -  Name: Mak
2024/07/21 12:00:54 Received Message: Subjects.AccountInfo
2024/07/21 12:00:54 Received Message Data: {
  "PageName": "Account_UI",
  "PolicyID": "1",
  "UpdatedTime": "Fri, 19 Jul 2024 23:54:04 EDT",
  "Element": {
    "Salutation": "",
    "FirstName": "Mak",
    "LastName": "",
    "MiddleName": "",
    "PrimaryAddr": {
      "Addr1": "1121",
      "Addr2": "",
      "City": "",
      "State": "",
      "Zip": "",
      "County": "",
      "Country": ""
    },
    "BillingAddr": {
      "Addr1": "",
      "Addr2": "",
      "City": "",
      "State": "",
      "Zip": "",
      "County": "",
      "Country": ""
    },
    "Relationships": {
      "Party": {
        "PartyID": "",
        "Type": "",
        "FirstName": "",
        "LastName": "",
        "MiddleName": ""
      }
    }
  }
}
2024/07/21 12:00:54 Consumer Data =>  Page: Account_UI  -  Name: Mak
2024/07/21 12:01:24 Received Message: Subjects.AccountInfo
2024/07/21 12:01:24 Received Message Data: {
  "PageName": "Account_UI",
  "PolicyID": "1",
  "UpdatedTime": "Thu, 11 Jul 2024 12:54:55 EDT",
  "Element": {
    "Salutation": "",
    "FirstName": "Mak",
    "LastName": "",
    "MiddleName": "",
    "PrimaryAddr": {
      "Addr1": "1121",
      "Addr2": "",
      "City": "",
      "State": "",
      "Zip": "",
      "County": "",
      "Country": ""
    },
    "BillingAddr": {
      "Addr1": "",
      "Addr2": "",
      "City": "",
      "State": "",
      "Zip": "",
      "County": "",
      "Country": ""
    },
    "Relationships": {
      "Party": {
        "PartyID": "",
        "Type": "",
        "FirstName": "",
        "LastName": "",
        "MiddleName": ""
      }
    }
  }
}

when I view in nats-cli.I see no new messages in this timestamp.Consumer code

package main

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"os/signal"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {

	nc, err := nats.Connect(nats.DefaultURL, nats.ReconnectWait(10*time.Second))

	if err != nil {
		log.Fatal(err)
	}
	defer nc.Drain()
	/*Drain is a safe way to ensure all buffered messages that were published
	are sent and all buffered messages received on a subscription are processed
	being closing the connection.*/

	ctx := context.Background()

	getStatusTxt := func(nc *nats.Conn) string {
		switch nc.Status() {
		case nats.CONNECTED:
			return "Connected"
		case nats.CLOSED:
			return "Closed"
		default:
			return "Other"
		}
	}
	log.Printf("The connection is %v\n", getStatusTxt(nc))

	// Use the JetStream context to produce and consumer messages that have been persisted.

	//js, err := nc.JetStream(nats.PublishAsyncMaxPending(10000)) //ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatal(err)
	}

	// Create a stream if it does not exist
	stream, err := js.Stream(ctx, "Subjects")
	if err != nil {
		log.Fatal(err)
	}

	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Name:      "Subjects",
		Durable:   "Subjects", //FilterSubject:Subjects.Account,AckPolicy:jetstream.AckExplicitPolicy
		AckPolicy: jetstream.AckExplicitPolicy,
	})

	if err != nil {
		log.Fatal(err)
	}

	ci, err := consumer.Info(ctx)
	if err != nil {
		log.Fatal(err)
	}

	log.Printf("Messages - Pending: %d, Waiting: %d, Ack Pending: %d", ci.NumPending, ci.NumWaiting, ci.NumAckPending)

	account := AccountInfoRecord{}
	cctx, err := consumer.Consume(func(msg jetstream.Msg) {
		log.Printf("Received Message: %s", string(msg.Subject()))
		log.Printf("Received Message Data: %s", string(msg.Data()))

		msg.Ack()
		err = json.Unmarshal([]byte(msg.Data()), &account)
		if err != nil {
			log.Fatal(err)
		}
		log.Printf("Consumer Data =>  Page: %s  -  Name: %s\n", account.PageName, account.Element.FirstName)
	})

	if err != nil {
		log.Fatal(err)
	}

	defer cctx.Stop()

	//gracefully shutdown
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, os.Interrupt)
	<-quit

}

[wallyqs]

In JS you can be delivered or redelivered messages without the producer being connected.