Consumer remains bound after unsubscribe?

[scottf]

Context: in the process of reproducing an issue, I have created a kotlin test to cycle through the following sequence;
create durable consumer on stream,
async subscribe to subject in consumer filter binding to consumer,
publish/receive,
unsubscribe async sub (via dispatcher)
delete consumer.
First run through works. Second run through fails with "[SUB-90012] Consumer is already bound to a subscription."
I assumed this meant that unsubscribe (step 4) had failed, and that deleting the durable consumer the subscription was bound (step 5) to have somehow survived to the second iteration.
Using the debugger, I also noticed, that when I unsubscribe using the dispatcher, sub returned is null.

NatsSubscription sub = this.subscriptionsUsingDefaultHandler.get(subject);
if (sub != null) {

Some other notes -
I did confirm using nats cli, that at the end of the first run, the consumer was indeed deleted:

$ nats consumer info foo_bar
nats: error: could not select Consumer: no Consumers are defined for Stream foo_bar

Calls to nats api in test look like this:

val jsManagement = conn?.jetStreamManagement(
	JetStreamOptions.Builder()
		.requestTimeout(Duration.ofSeconds(5))
		.build()
)

for (i in 1..100) {
	// Create our durable consumer
	val consumerInfo = jsManagement?.addOrUpdateConsumer(
		stream,
		ConsumerConfiguration.Builder()
			.durable(consumerName)
			.deliverSubject(deliverSubject)
			.filterSubject(subject)
			.ackPolicy(AckPolicy.Explicit)
			.build()
	)

	var count = AtomicInteger()
	val d = conn?.createDispatcher()
	val h = MessageHandler { msg ->
		Log.d(
			TAG,
			"Received message on thread: ${Looper.myLooper()}/${Thread.currentThread()}"
		)
		count.getAndIncrement()
	}

	val so = PushSubscribeOptions.bind(stream, consumerName)
	val sub = js!!.subscribe(subject, d, h, /* autoAck= */ true, so)
	Log.i(
		TAG,
		"Successful subscription: $subject on thread: ${Looper.myLooper()}/${Thread.currentThread()}"
	)
	Assert.assertNotNull(sub)
	Assert.assertEquals(0, count.toInt())

	val msg = byteArrayOfInts(0xA1, 0x2E, 0x38, 0xD4, 0x89, 0xC3)
	js?.publish(
		subject, msg,
		PublishOptions.Builder()
			.streamTimeout(Duration.ofSeconds(1))
			.build()
	)

	val testFn: () -> Boolean = {
		Log.d(TAG, "COUNT: ${count.toInt()}")
		count.toInt() > 0
	}
	Assert.assertTrue(waitFor(200, Duration.ofMillis(10), testFn))
	d!!.unsubscribe(sub.subject)

	Assert.assertTrue(jsManagement!!.deleteConsumer(stream, consumerName))
}

[scottf]

Seems like a timing thing. How long between issuing the delete and the re-issue of the sub? There might be some lag amount of time that it takes for the server to remove the consumer. Also, why are you doing this, what is the use case? You could always make a unique durable name each iteration

[scottf]

[Christian Cassar]
I am trying to reproduce an issue. I initially thought it was a timing thing, but I tried with a delay at the end of the loop (30s) and issue persists. Conversely, the first run succeeds irrepective (so maybe somehow also tied to client connection lifetime? Will experiment some more.

(Also I am a little surprised that my call to dispatcher unsubscribe returns null at the point indicated above - makes me wonder if maybe I am calling the wrong api?)

[scottf]

[Christian Cassar] An extra datapoint here - if I tear down client connection, and reestablish it with every iteration, issue goes away.

I am trying to reproduce an issue. I initially thought it was a timing thing, but I tried with a delay at the end of the loop (30s) and issue persists. Conversely, the first run succeeds irrespective (so maybe somehow also tied to client connection lifetime? Will experiment some more.

[ccassar]

For completeness:

I updated my test script (still in the process of trying to repro original issue), I changed the sequence to disconnect and reconnect between iterations AND to create a new consumer with every iteration and delete it at the end (but NOT reuse it on subsequent iterations).

After a number of iterations (in this case 18K+) consumer deletion returned an error, but was obviously still deleted:

rawtestclient: Received expected index 18436 subject foo.bar.baz consumer rawtestclient18436 stream foo_bar ds acme.deliver_subject.rawtestclient.foo.bar.baz on thread: null/Thread[rawtestclient:13,5,main]
rawtestclient: consumer deletion failed: io.nats.client.JetStreamApiException: unlinkat /tmp/nats/jetstream/$G/streams/foo_bar/obs/rawtestclient18436: directory not empty [10051], will retry
rawtestclient: consumer deletion failed: io.nats.client.JetStreamApiException: consumer not found [10014], will retry

code:

            // Create our durable consumer
            jsManagement?.addOrUpdateConsumer(
                stream,
                ConsumerConfiguration.Builder()
                    .durable(durableCN)
                    .deliverSubject(deliverSubject)
                    .filterSubject(subject)
                    .ackPolicy(AckPolicy.Explicit)
                    .build()
            )
...
            do {
                try {
                    deleted = jsManagement!!.deleteConsumer(stream, durableCN)
                } catch (e: Exception) {
                    failedConsumerDelete++
                    Log.i(TAG, "consumer deletion failed: ${e}, will retry")
                    Thread.sleep(Duration.ofSeconds(30).toMillis())
                }
            } while (!deleted)

In summary, a couple of issues so far;

1. if I do not disconnect/reconnect, and simply delete durable consumer and readd durable consumer I consistently fail to subscribe because consumer is already associated with push.
2. if I do disconnect after some number of iterations, deletion of consumer sometime appears to fail, but hasn't.

I can work around both of these so far but thought I should record the behaviour here in case it is pertinent to others.