I am fetching couchbase mutations through java code similar to run function in https://github.com/couchbase/couchbase-kafka-connector/blob/master/src/main/java/com/couchbase/kafka/CouchbaseReader.java . We subscribe with a io scheduler instead of toBlocking at the end. We get mutations and streamEndMessages but we never get onComplete. Hence the code just waits until killed. We use core-io-1.2.6, and with top of branch core-io-1.3.0.
This is known limitation at the moment, which will be fixed in next version