There is a much simpler way of fixing this: undo all these latest changes and request 1!

if (BackpressureUtils.getAndAddRequest(this, 1) == 0) {
    parent.onUnsubscribe(state);
}

It's interesting. If there is currently a thread processing over the request count loop (i.e. processing nextIteration()) then won't this have to block the unsubscribing thread? The unsubscribing thread would have to poll over this until it got access. I'm not seeing how this would force a thread currently iterating to terminate. If you read through the concurrency I think you will see that it's not really all that complicated and I believe it sufficiently covers the possible race conditions.

There is no blocking. My suggestion follows the queue-drain approach where you queue the terminal flag and drain it in the emission loop. Since the loop first checks for the terminal flag there won't be any wrong emissions and the loop quits. If unsubscribe increments from zero, there isn't and won't be any emission loop after that and the resource can be freed.

There is no need for your complicated solution.

I'm not seeing where you are suggesting setting a terminal flag. You are more than welcome to open your own pull request to demonstrate your fix. My pull request does set a flag for the draining thread to signal when it needs to unsubscribe.