title: OOM issue while copying couchbase documents from one bucket to another bucket using reactive programming
body: <p>We are trying to copy data from one bucket to another using reactive programming(around 1 million documents). We are getting OOM in this piece of code. I am no rxjava expert and need help in order to prevent OOM. I think the read is happening faster than the write and that’s causing the OOM due to buffer getting full. The code is as follows:</p>
  
  <pre><code>CountDownLatch countDownLatch5 = new CountDownLatch(1);
  Observable
          .from(n1qlKeysForDocsGPC)
          .flatMap(new Func1&lt;String, Observable&lt;JsonDocument&gt;&gt;() {
              @Override
              public Observable&lt;JsonDocument&gt; call(String key) {
                  return readPrimaryMainAsyncBucket
                          .get(key, 10, TimeUnit.SECONDS)
                          .onErrorResumeNext(readPrimaryMainAsyncBucket.get(key, 10, TimeUnit.SECONDS))
                          .retry(50)
                          .switchIfEmpty(Observable.empty())
                          .onErrorResumeNext(Observable.empty());
              }
          })
          .flatMap(new Func1&lt;JsonDocument, Observable&lt;JsonDocument&gt;&gt;() {
              @Override
              public Observable&lt;JsonDocument&gt; call(JsonDocument jsonDocument) {
                  return readPrimaryBackupAsyncBucket.upsert(jsonDocument, 10, TimeUnit.SECONDS).retry(50);
              }
          })
          .last()
          .doOnTerminate(new Action0() {
              @Override
              public void call() {
                  countDownLatch5.countDown();
              }
          })
          .subscribe();
  try {
      countDownLatch5.await();
      logger.info("DataRecoverySchedulers | Completed countDownLatch5");
  } catch (InterruptedException e) {
      e.printStackTrace();
  }
  </code></pre>