Fix zip race condition

[vadims]

The ItemObserver collection logic needs to run in both the OnNext and OnCompleted handlers.

Otherwise, you might have values waiting to be emitted and without seeing an OnCompleted. I couldn't think of an easy unit test for this, it actually took a while to narrow down the example.

The only way I could reproduce it on my machine was to use Hystrix and it happens somewhat sporadically (10-50 iterations).

public class Test {

  public static class DummyCommand extends HystrixCommand<Integer> {

    protected DummyCommand() {
      super(HystrixCommandGroupKey.Factory.asKey("Dummy"));
    }

    @Override
    protected Integer run() throws Exception {
      Thread.sleep(1);
      return 1;
    }
  }

  public static void main(String[] args) throws InterruptedException {

    while (true) {
      Observable.zip(tap(new DummyCommand().observe()), tap(new DummyCommand().observe()), new Func2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer a, Integer b) {
          return a + b;
        }
      }).toBlockingObservable().single();
      System.out.println("\n\n\n");
      Thread.sleep(5);
    }

  }

  public static Observable<Integer> tap(Observable<Integer> observable) {
    return observable.doOnEach(new Action1<Notification<? super Integer>>() {
      @Override
      public void call(Notification<? super Integer> notification) {
        System.out.println("notification = " + notification);
      }
    });
  }

}

[cloudbees-pull-request-builder]

RxJava-pull-requests #692 FAILURE
Looks like there's a problem with this pull request

[vadims]

Not sure why it failed, since it passes locally and zip is not used in that test case. Is there a way to re-trigger the build?

[akarnokd]

Nice catch. Basically, you need something like T1:onNext, (T1: onCompleted, T2: onNext) T2: onCompleted where T1 onCompleted wins the lock over T2: onNext, then T2 onCompleted just doens't do anything as the queues aren't empty. I confirm your fix solves this issue.

We have a few unreliable tests and rerunning might not produce success. Closing and reopening the PR might trigger a run, but I'm not sure.

[vadims]

@benjchristensen any idea on how to re-trigger CI build?

[benjchristensen]

any idea on how to re-trigger CI build?

Only way I know how is updating the PR such as with a new commit :-( It's quite annoying that way.

[benjchristensen]

Is it possible to demonstrate the race with a unit test, perhaps using latches to cause the necessary race?

Reason is that the zip operator will be changing soon to use the new design changes and I want to make sure we have a test to prove correctness.

[vadims]

Unfortunately it's an internal race between acquiring the read lock and trying to acquire the write lock. I don't have any good ideas on how to test it without coupling it to the specific implementation.

[cloudbees-pull-request-builder]

RxJava-pull-requests #693 FAILURE
Looks like there's a problem with this pull request

[cloudbees-pull-request-builder]

RxJava-pull-requests #694 SUCCESS
This pull request looks good