Can elastic pcgroups be an InterestPolicy source stream?

[nickchomey]

InterestPolicy streams suit my use case perfectly, but I'd also like to use elastic Partitioned Consumer Groups. However, elastic pcgroups seem to create a WorkQueuePolicy.

Might it be possible to create an InterestPolicy instead? Or would that not be compatible with the guaranteed-order, deliver-once, multi-member nature of pcgroups?

[jnmoyne]

Elastic PCG uses a work queue stream as an 'intermediary' stream (that sources the stream where the data is stored), where the partition number is inserted into the subjects and where the messages are deleted (from that intermediary stream) as they are consumed. Interest policy would not be very very suited to this because with interest if you do not have a durable consumer defined that matches the subject of the message when it gets stored into the stream then the message doesn't get stored in the stream. You would not want this to happen on this stream because it means if for example there is at the time the message is sourced into the intermediary stream no member actively creating a consumer for the partition numbers that are assigned to it then the message would just 'disappear' and this would lead to the consumer group appearing to be 'losing messages'.

Also, you can't source from an interest stream, and in any case the intermediary would want to source every message in the stream, and only once so even if this was allowed it would be no different than sourcing from a work queue stream (which, you also cannot do at the moment).

What you could do however is use the 'static' PCG instead, which doesn't use an intermediary stream and creates consumers directly on the stream where the data is stored, so that stream could have any retention policy you would want (including work queue and interest). So for example insert the partition number (i.e. the max number of members you think you would ever need) into the messages as they get stored in the stream (easily done using subject transformation as part of the stream's config), and then you can still get 'elasticity' if that stream is set to work queue, to change the list of members for the static consumer group just delete the consumer group (this will delete the existing consumers for the PCG) and then re-create the static PCG with a different membership list. If indeed the stream is set as a work queue, the messages will get deleted from the stream as they are consumed, meaning that when you delete the PCG and create a new one with a different membership list then it will just pick up where it had left of with the previous instance of the PCG. This effectively gives you the ability to have something equivalent to 'elastic' (but using 'static' and a work queue stream).

[nickchomey]

Thanks for the thoughtful reply!

My hope was to be able to combine the ability of InterestPolicy to drop messages to a subject that doesn't have a consumer, with the high-availability strict ordering and partitioning of pcgroups. But, yeah, I think its just not possible.

I tried changing line 303 in the test to InterestPolicy, but that failed. Perhaps it has something to do with there being multiple consumers thus the message not getting dropped and thus double-counted.

orbit.go/pcgroups/elastic_stream_consumer_group.go

Lines 301 to 324 in a5d768f

	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
	Name: composeCGSName(streamName, consumerGroupName),
	Retention: jetstream.WorkQueuePolicy,
	Replicas: replicas,
	Storage: storage,
	MaxMsgs: maxBufferedMessages,
	MaxBytes: maxBufferedBytes,
	Discard: jetstream.DiscardNew,
	Sources: []*jetstream.StreamSource{
	{
	Name: streamName,
	OptStartSeq: 0,
	OptStartTime: nil,
	FilterSubject: "",
	SubjectTransforms: []jetstream.SubjectTransformConfig{
	{
	Source: filter,
	Destination: filterDest,
	},
	},
	},
	},
	AllowDirect: true,
	})

Anyway, I'll think about perhaps trying to implement something with a static consumer group. But the elastic really is what I was looking for.

Thanks for your time!