3.x: Fix window (boundary, start/end) cancel and abandonment (#6762)

Author: akarnokd

src/main/java/io/reactivex/rxjava3/core/Flowable.java

@@ -17965,6 +17965,11 @@ public final Flowable> window(
      * Publisher.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Backpressure:
      *  The outer Publisher of this operator does not support backpressure as it uses a {@code boundary} Publisher to control data
@@ -17995,6 +18000,11 @@ public final  Flowable> window(Publisher boundaryIndicator) {
      * Publisher.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Backpressure:
      *  The outer Publisher of this operator does not support backpressure as it uses a {@code boundary} Publisher to control data
@@ -18031,6 +18041,11 @@ public final  Flowable> window(Publisher boundaryIndicator, in
      * {@code closingSelector} emits an item.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Backpressure:
      *  The outer Publisher of this operator doesn't support backpressure because the emission of new
@@ -18068,6 +18083,11 @@ public final  Flowable> window(
      * {@code closingSelector} emits an item.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Backpressure:
      *  The outer Publisher of this operator doesn't support backpressure because the emission of new

src/main/java/io/reactivex/rxjava3/core/Observable.java

@@ -14536,6 +14536,11 @@ public final Observable> window(long count, long skip, int bufferS
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} operates by default on the {@code computation} {@link Scheduler}.
@@ -14564,6 +14569,11 @@ public final Observable> window(long timespan, long timeskip, Time
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14594,6 +14604,11 @@ public final Observable> window(long timespan, long timeskip, Time
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14630,6 +14645,11 @@ public final Observable> window(long timespan, long timeskip, Time
      * ObservableSource emits the current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} operates by default on the {@code computation} {@link Scheduler}.
@@ -14658,6 +14678,11 @@ public final Observable> window(long timespan, TimeUnit unit) {
      * emits the current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} operates by default on the {@code computation} {@link Scheduler}.
@@ -14690,6 +14715,11 @@ public final Observable> window(long timespan, TimeUnit unit,
      * emits the current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} operates by default on the {@code computation} {@link Scheduler}.
@@ -14723,6 +14753,11 @@ public final Observable> window(long timespan, TimeUnit unit,
      * ObservableSource emits the current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14754,6 +14789,11 @@ public final Observable> window(long timespan, TimeUnit unit,
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14788,6 +14828,11 @@ public final Observable> window(long timespan, TimeUnit unit,
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14824,6 +14869,11 @@ public final Observable> window(long timespan, TimeUnit unit,
      * current window and propagates the notification from the source ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  You specify which {@link Scheduler} this operator will use.
@@ -14865,6 +14915,11 @@ public final Observable> window(
      * ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} does not operate by default on a particular {@link Scheduler}.
@@ -14891,6 +14946,11 @@ public final  Observable> window(ObservableSource boundary)
      * ObservableSource.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} does not operate by default on a particular {@link Scheduler}.
@@ -14922,6 +14982,11 @@ public final  Observable> window(ObservableSource boundary,
      * {@code closingIndicator} emits an item.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} does not operate by default on a particular {@link Scheduler}.
@@ -14953,6 +15018,11 @@ public final  Observable> window(
      * {@code closingIndicator} emits an item.
      * 
      * 
+     * 
+     * Note that ignoring windows or subscribing later (i.e., on another thread) will result in
+     * so-called window abandonment where a window may not contain any elements. In this case, subsequent
+     * elements will be dropped until the condition for the next window boundary is satisfied. The behavior is
+     * a tradeoff for ensuring upstream cancellation can happen under some race conditions.
      * 
      *  Scheduler:
      *  This version of {@code window} does not operate by default on a particular {@link Scheduler}.

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableWindowBoundary.java

@@ -240,7 +240,11 @@ void drain() {
 
                         if (emitted != requested.get()) {
                             emitted++;
-                            downstream.onNext(w);
+                            FlowableWindowSubscribeIntercept<T> intercept = new FlowableWindowSubscribeIntercept<T>(w);
+                            downstream.onNext(intercept);
+                            if (intercept.tryAbandon()) {
+                                w.onComplete();
+                            }
                         } else {
                             SubscriptionHelper.cancel(upstream);
                             boundarySubscriber.dispose();