Empty hot source hangs with 2nd late subscriber about reactor-core

[chemicL]

Hi, @RonBarkan 👋

Please consider the following:

@Test
void lateSuscribe() throws Exception {
	CountDownLatch latch = new CountDownLatch(1);

	ConnectableFlux<Integer> publish = Flux.just(1)
	                                       .publish();

	publish.subscribe(
			r -> System.out.println("1.Next: " + r), 
			e -> System.out.println("1.Error " + e),
			() -> System.out.println("1.Done"));

	publish.connect();

	publish.doFinally(s -> latch.countDown())
	       .subscribe(r -> System.out.println("2.Next: " + r),
			       e -> System.out.println("2.Error " + e),
			       () -> System.out.println("2.Done"));

	latch.await();
}

And the output:

1.Next: 1
1.Done

It also does not finish. That is because, in my view, late-arriving Subscriber has no defined contract in case of publish().
Please consider using replay() to achieve the desired result. Please consult our reference documentation for some examples.

With the above, I'm closing this issue. For further questions around such behaviour, StackOverflow might be a good place. For instance, I found a similar question with a reasonable answer: https://stackoverflow.com/questions/59237587/reactor-publish-behavior

[RonBarkan]

I'd expect the 2nd subscriber to get the complete signal. If the published flux had elements, the late subscriber should not see them and only get the complete signal.

This is what I understand from the marble diagram in the JavaDoc of Flux.publish(). The link you provided does not say anything about late subscribers, AFAICT. Consider updating the docs.
Lastly, the stackoverflow answer does exactly what I described above, and does not discuss hanging.

In any case, it seems a very weird desired behavior to have pipelines hang for this possibly subtle issue. Seems to me that even throwing an exception will be preferred. In particular, the published flux accepted the subscription and then ignored it. It is certainly a surprise to see your pipeline hang with the right set of inputs and it should be a design goal to reduce API usage surprises.

[RonBarkan]

(Regarding replay() thank you for your suggestion. Even though the original pipeline had a significant flaw, not related to this issue, both publish() and replay() / cache() are not appropriate. The real code is being change to do something else.)

[chemicL]

Thanks for providing more feedback @RonBarkan. I did a bit of digging in the history and found #2897 - it feels it's the same subject, so I marked this as duplicate. It also contains the suggestion of using replay(history=0) to overcome the limitations of publish. After giving it a second thought, it does feel unexpected to have a hanging process and coordinating anything following a completion to avoid the late subscription feels undesired. I can't tell much about the priorities and timeline around this, as there are other ongoing efforts, but a contribution is always welcome. At the same time, having a workaround takes away the pressure I suppose.