Ephemeral Consumer Cleanup Rate?

[alexang-zh]

Hi, we have a use case for ephemeral consumers where we would need to create them per API call we receive at a relatively large RPS rate. We create an ephemeral consumer with an inactivity threshold of 1 minute for the purposes of fetching up to the latest 100 messages from the subject.

However, the number of consumers still keep growing (> 1million consumers) despite having set that value.

What is the rate at which ephemeral consumers are deleted? Is this a valid use case for ephemeral consumers?

Example usage:

func (r *repository) GetUserData(
	ctx context.Context,
	userCode string,
) ([]string, error) {
	consumerName := fmt.Sprintf(
		"fetch-user-data-%s-%s",
		userCode,
		uuid.NewString(),
	)
	cs, err := r.js.CreateConsumer(
		ctx,
		"some stream",
		natsJetstream.ConsumerConfig{
			Name:              consumerName,
			Durable:           "", // This is not a durable consumer, DO NOT SET.
			Description:       "some description.",
			DeliverPolicy:     natsJetstream.DeliverAllPolicy,
			AckPolicy:         natsJetstream.AckNonePolicy,
			FilterSubject:     fmt.Sprintf("USER_DATA.%s", userCode),
			ReplayPolicy:      natsJetstream.ReplayInstantPolicy,
			MaxRequestBatch:   100,
			InactiveThreshold: NATSInactiveThreshold, // time.minute
		},
	)
	if err != nil {
		return nil, errors.Wrap(err, "failed to initialize consumer")
	}

	res, err := cs.FetchNoWait(100)
	if err != nil {
		return nil, errors.Wrap(err, "fetch failed")
	}

	for msg := range res.Messages() {
		// ... message processing
		fmt.Println(msg)
	}

	return nil, nil
}

[derekcollison]

What server version? What is the topology of the NATS system?

[alexang-zh]

version: 2.10.3

NATS cluster with 5 nodes running on ec2, one server instance per node, c7gn.2xlarge instances

[derekcollison]

Please upgrade to latest server, 2.10.9. If the issue persists, please let us know. Note that creating and throwing away lots of consumers though is a concern that would be good to understand what you are trying to achieve.

[alexang-zh]

We are trying to set up NATS as our persistence layer and means to query it.

When querying for a user's history, we'd create a consumer to fetch the messages for that user from NATS via the user ID as a filter subject. User state is published to a NATS subject akin to USER.<user_id> each time a user is added, updated or deleted.

The messages in the response would represent the full historical states of the user.

[ripienaar]

Such a high rate of consumer creation will not really work well. Would be better to look into the APIs we have to fetch single messages from the stream - we are about to land a mode that would allow batches to be fetched without a consumer. See nats-io/nats-architecture-and-design#262

[alexang-zh]

Thanks @ripienaar, that would be ideal for our use case. Are there be any plans to add that to the nats.go jetstream simplified library's Stream interface?

nats.go/jetstream/stream.go

Line 30 in f42d2be

Stream interface {

[ripienaar]

Yes, for sure, once this ships expect it in the clients.

[alexang-zh]

Sorry, last question - any ETA to the above? So we can decide if we should at least refactor to fit the history in a single message and use GetLastMessageForSubject as a near term solution.

[ripienaar]

If you can compact the history to a single key that's of course best, streams are optimised for tail of data not random access, so regardless of when we ship this if that's possible I would recommend it. We have Rollup support in streams to make this better and KV buckets use that quite a bit.

It should be in next server release in a few weeks and clients soon after - go tends to be first but things can change so don't take that as gospel :)

[derekcollison]

In addition to what @ripienaar has shared.

For using consumers today, make sure the consumer to fetch the history is an R1 and memory backed, that can help or an ordered consumer which will be an R1 and flow controlled.

Also, even prior toi batch support landing (it is already on main and nightlies for the server), you could use NextFor and walk the user history today by issuing a new request with a different starting seq upon receipt of the prior response. Then when batch lands you could simply add in Batch to get more than one at a time.

[alexang-zh]

Hi @derekcollison @ripienaar,

I have a follow up question that's adjacently related.

We have a requirement to implement an optimistic realtime streaming architecture that loops back onto itself:

• Our client hits our API
• We publish the message onto first-level stream, but do not yet return a response to the client
• Our downstream processor consumes the message from the first-level stream and publishes it onto second-level stream
• The same API call receives the result of the second-level event, then returns an appropriate response back to the client

Is this architecture not possible with JetStream or NATS? How would you recommend building for this requirement?

We do not need ack/nak support like a message consumer for this as it is optimistic, so long as we get a realtime update that it has been processed by our downstream consumer, it is sufficient. Will spawning a NATS subscriber (instead of a consumer) on every API call be ok for scaling?

[Jarema]

First of all, Container B will get ack for publishing into NATS Stream A without any consumer needed to be created. Jetstream utilizes requet/response pattern for that. It is very efficient, and do not use consumers at all. You wil receive ack for every jetstream.Publish

Secondly, yes. You can get ack from Stream B.
There are two ways of doing that:

1. Create a single longer living consumer that would process all messages stored in Stream B and use it on Container B. That could be a consumer with a wildcard.
2. use Republish feature on Steam B, that will republish given messages to another subject, on which you could subscribe with just Core Nats subscription.

[alexang-zh]

@Jarema For option 2, are there any scaling issues if there is a short-lived subscription (<30s) on every API call?

I've been exploring option 1 - but it seems that as we scale, there will be proportionately increased latency from having to filter out messages and route them back to the same thread handling the API call. (imagine having to sift through 99% of other API calls' messages when only looking for that 1)