Couchbase as an Idempotent Repository

[behrad]

I want to implement an Idempotent Repository on Couchbase with three synchronous add, contains, remove operations, which will be called concurrently in a very high rate (5000 calls/sec each)

my very simple implementation is as below:

//Remove Op
val res = Await.result( bucket.delete( key ), 1 second )

// Contains check
val res = Await.result( bucket.get[String]( key ), 1 second )
res.exists( _ => true )

// Add operation
val res = Await.result( bucket.add( key, System.currentTimeMillis, CouchbaseExpirationTiming_byDuration( expiry ) ), 1 second )
res.isSuccess

however this is not scalable and my operations time out under load.
How can I achieve better implementation? can stream methods help me with sync operations needed above?

[mathieuancelin]

Just one question, why do you use Await ?

[mathieuancelin]

Also, I'm not sure to understand what you are trying to achieve.
Some kind of replace operation ?

[behrad]

I am implementing Camels IdempotentRepositorys SPI here: https://github.com/apache/camel/blob/master/camel-core/src/main/java/org/apache/camel/spi/IdempotentRepository.java

[mathieuancelin]

Hum, okay I see.

The problem here is the 1 second timeout, you can't rely on that under pressure.

[behrad]

then turning Await to a non blocking async operation wont help then...
If i can't rely on many concurrent get/set operations under pressure, what other couchbase operation could feed me here? what about streamset... !?

[mathieuancelin]

As you seems to need blocking operations, either you can wait longer, of you can use the underlying driver and access blocking operations.

[chenbekor]

hi @behrad sorry for jumping late to the conversation ;) I am not familiar with Camel can you please describe in more details what is the exact use case? add, contains, remove operations running concurrently on the same document / key? 5000 calls/sec - is this concurrent updates on the same entity?

from past experience 5000/sec should not be an issue for couchbase. what scala framework are you using to handle 5K/sec load?

[behrad]

@chenbekor that Camel interface is an IdempotentRepository which is to ensure no duplicate message is processed in Camel. Each message arriving gets a UUID and is added to a couchbase bucket via that add method add(uuid) when processed by Camel, we are using akka atLeastOnceDelivery so that messages are retried if no ack received at the end of message processing. however we need a exactly once delivery semantics, so we need idempotency at the message processing so that it skips duplicates messages (contains method). successfully processed messages are removed (remove method)
here is how camel is doing this: https://github.com/apache/camel/blob/master/camel-core/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java

So I can say that, these operations are called 90% on new uuids, we only want to catch duplicate message processing on some exceptional use cases.

You can see a simple in memory LRU implementation of that idempotent cache here:
https://github.com/apache/camel/blob/master/camel-core/src/main/java/org/apache/camel/processor/idempotent/MemoryIdempotentRepository.java
however this is not distributed and scalable :)

Couchbase would be a nice candidate for this scenario I thought...

[chenbekor]

I'm trying to help so please excuse me for asking you to provide more details :)

Let's try again - but this time, if you can avoid the Camel related logic and also - if you can omit the Idempotent repository term.

Try to describe in very simple words what it is that you are looking from couch. Example:

I want to set a document in one thread and ensure that when I'm trying to access the same doc from a different thread I will get "key already exist"

or maybe:

I want to set from one thread and read concurrently from another thread etc

this will help me allot to understand what you are trying to achieve.

[behrad]

I expect couchbase as a distributed in memory hash table, and I put message ids as keys there.
for each message I want to ensure that key doesn't exist so I process it. if it exists I cancel that message :)

Thank you for your attention and time @chenbekor

[behrad]

I think I can move these checks to some other place where there is possible to bulkGet ids to filter exist ones and bulkSet to set new ones. this is more performant.
however I don't know how to translate this into reactivecouchbase or scala:

public List<JsonDocument> bulkGet(final Collection<String> ids) {
    return Observable
        .from(ids)
        .flatMap(new Func1<String, Observable<JsonDocument>>() {
            @Override
            public Observable<JsonDocument> call(String id) {
                return bucket.async().get(id);
            }
        })
        .toList()
        .toBlocking()
        .single();
}

[behrad]

may be off topic: Can you also provide me a code snip show how to use stream api in reactivecouchbase? could they help me here?

[mathieuancelin]

As you seems to need blocking operations (because of Camel), streams will not help you there.

So I guess your first instinct was right, only the timeout was wrong

object Repo extends Service[String] {

  import org.reactivecouchbase.CouchbaseRWImplicits.jsValueToDocumentWriter
  import org.reactivecouchbase.CouchbaseRWImplicits.documentAsJsValuetReader

  private[this] val maxDuration = Duration(Long.MaxValue, TimeUnit.MILLISECONDS)
  private[this] val driver = ReactiveCouchbaseDriver()
  private[this] val bucket = driver.bucket("default")

  override def add(key: String): Boolean = {
    Await.result(bucket.add[JsNumber](key, JsNumber(System.currentTimeMillis())).map(_.isSuccess), maxDuration)
  }
  override def contains(key: String): Boolean = {
    Await.result(bucket.get[JsNumber](key).map(_.isDefined), maxDuration)
  }
  override def remove(key: String): Boolean = {
    Await.result(bucket.delete(key).map(_.isSuccess), maxDuration)
  }
  override def confirm(key: String): Boolean = ???  // I don't understand the meaning of that method
}

and the Scala way for your previous piece of code is something like

def bulkGet(ids: List[String]): List[JsObject] = {
  import org.reactivecouchbase.CouchbaseRWImplicits.documentAsJsObjectReader
  Await.result(   // block until future is completed
    Future.sequence(  // transform a List[Future[M]] to Future[List[M]]
      ids.map(id => bucket.get[JsObject](id)) // for each id fetch the document in couchbase
        .map { optDocsList =>
          optDocsList.collect {   // as a document may not be present for an id, filter all None options
            case Some(doc) => doc
          }
        }
    ),
    Duration(Long.MaxValue, TimeUnit.MILLISECONDS)    // wait forever
  )
}

[chenbekor]

I think that Mathiieu's code is always a real educational stuff - again Mathieu thank you for all that you are doing here.

@behrad here is my opinion in terms of scalable solution that answers your requirement.

What I would do here is wrap the couch related functionality in a separate micro service. I would expose the service via a REST API (or more performant protocols only if a bottleneck is identified)

Next, I would put a load balancer in front of this micro service. If you are running on AWS I would recommend using ELB and Auto Scaling Group. We see great results with this approach.

Now - your micro service can scale in / out to handle thousands of concurrent requests.

To be more pragmatic I would recommend play framework which has great integration with reactive couchbase thanks to Mathieu. This will allow you to utilize the async API and the non-blocking nature of Play while retaining full synchronous operation from a REST API standpoint.

So internally - the service is async/non blocking but externally you provide a synchronous API which is exactly what you need for your use case. You can now call the REST API from your other parts (I guess the Camel plugin) and block on the micro service callback.

needless to say - play framework can be replaced by spray or finagle or what have you.

Couchbase is rarely a bottleneck from our experience.