Q: RxJava: Merging Observable with Completable does not work I have an Observable that at some point has to write things to the cache - and we would like to wait that writes are done before finishing the whole operation on the observable (for reporting purposes).
For the purpose of test, the cache write Completable looks like this:
   Completable.create(
                  emitter ->
                      new Thread(
                              () -> {
                                try {
                                  Thread.sleep(2000);
                                  doSomething();
                                  emitter.onComplete();
                                } catch (InterruptedException e) {
                                  e.printStackTrace();
                                }
                              })
                          .start());

Since I have several cache writes, I try to merge them in a container class:
public class CacheInsertionResultsTracker {

  private Completable cacheInsertResultsCompletable;

  public CacheInsertionResultsTracker() {
    this.cacheInsertResultsCompletable = Completable.complete();
  }

  public synchronized void add(Completable cacheInsertResult) {
    this.cacheInsertResultsCompletable = this.cacheInsertResultsCompletable.mergeWith(cacheInsertResult);
  }

  public Completable getCompletable() {
    return this.cacheInsertResultsCompletable;
  }
}

And I try to merge it with Observable in a following way:
CacheInsertionResultsTracker tracker = new ...;
    observable
        .doOnNext(next->tracker.add(next.writeToCache(...)))
        .mergeWith(Completable.defer(()->tracker.getCompletable()))
        .subscribe(
            // on next
            this::logNextElement
            // on error
            this::finishWithError
            // on complete
            this::finishWithSuccess
            );

How could I make sure that by the time finishWithSuccess is called the doSomething is completed?
The problem is that the Completable reference is updated every time I add a new one, and it happens after the mergeWith runs...

A: The solution that seems to work for our use case is to use concatWith + defer:
observable
    .doOnNext(next->tracker.add(next.writeToCache(...)))
    .concatWith(Completable.defer(()->tracker.getCompletable()))
    .subscribe(
        // on next
        this::logNextElement
        // on error
        this::finishWithError
        // on complete
        this::finishWithSuccess
        );

Concat assures that the subscription to the Completable happens only after the Observable is done, and defer defers getting the final Completable till this subscription (so all the objects are already added to the tracker).

A: Based on the comments, you could replace the completable cache with ReplaySubject<Completable>, do some timeout to detect inactivity and have the observable sequence end.
 ReplaySubject<Completable> cache = ReplaySubject.create();

 cache.onNext(completable);

 observable.mergeWith(
     cache.flatMapCompletable(v -> v)
          .timeout(10, TimeUnit.MILLISECONDS, Completable.complete())
 )

Edit:
Your updated example implies you want to run Completables in response to items in the main observable, isolated to that sequence, and wait for all of them to complete. This is a typical use case for flatMap:
  observable.flatMap(
       next -> next.writeToCache(...).andThen(Observable.just(next))
  )
  .subscribe(
        this::logNextElement
        // on error
        this::finishWithError
        // on complete
        this::finishWithSuccess
   );

