Couchbase

Apache Pekko Connectors Couchbase allows you to read and write to Couchbase. You can query a bucket from CouchbaseSource using N1QL queries or reading by document ID. Couchbase connector uses Couchbase Java SDK version 2.7.16 behind the scenes.

The Couchbase connector supports all document formats which are supported by the SDK. All those formats use the Document<T>Document[T] interface and this is the level of abstraction that this connector is using.

Project Info: Apache Pekko Connectors Couchbase
Artifact	org.apache.pekko pekko-connectors-couchbase 1.0.2 Snapshots are available
JDK versions	OpenJDK 8 OpenJDK 11 OpenJDK 17
Scala versions	2.13.13, 2.12.19, 3.3.3
JPMS module name	pekko.stream.connectors.couchbase
License	Apache-2.0
API documentation	API (Scaladoc)
Forums	Github Discussions
Release notes	GitHub releases
Issues	Github issues
Sources	https://github.com/apache/pekko-connectors

Artifacts

sbt

val PekkoVersion = "1.0.2"
libraryDependencies ++= Seq(
  "org.apache.pekko" %% "pekko-connectors-couchbase" % "1.0.2",
  "org.apache.pekko" %% "pekko-stream" % PekkoVersion
)

Maven

<properties>
  <pekko.version>1.0.2</pekko.version>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-connectors-couchbase_${scala.binary.version}</artifactId>
    <version>1.0.2</version>
  </dependency>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-stream_${scala.binary.version}</artifactId>
    <version>${pekko.version}</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  PekkoVersion: "1.0.2",
  ScalaBinary: "2.13"
]
dependencies {
  implementation "org.apache.pekko:pekko-connectors-couchbase_${versions.ScalaBinary}:1.0.2"
  implementation "org.apache.pekko:pekko-stream_${versions.ScalaBinary}:${versions.PekkoVersion}"
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Overview

Apache Pekko Connectors Couchbase offers both Apache Pekko Streams APIs and a more direct API to access Couchbase:

• CouchbaseSessionCouchbaseSession offers a direct API for one-off operations
• CouchbaseSessionRegistryCouchbaseSessionRegistry is an Apache Pekko extension to keep track and share CouchbaseSessions within an ActorSystem
• CouchbaseSourceCouchbaseSource, CouchbaseFlowCouchbaseFlow, and CouchbaseSinkCouchbaseSink offer factory methods to create Apache Pekko Stream operators

Configuration

All operations use the CouchbaseSession internally. A session is configured with CouchbaseSessionSettingsCouchbaseSessionSettings and a Couchbase bucket name. The Apache Pekko Stream factory methods create and access the corresponding session instance behind the scenes.

By default the CouchbaseSessionSettings are read from the pekko.connectors.couchbase.session section from the configuration eg. in your application.conf.

Settings

copysourcepekko.connectors.couchbase {
  session {
    nodes = ["localhost"]
    username = "Administrator"
    password = "password"
  }
}

Using Apache Pekko Discovery

To delegate the configuration of Couchbase nodes to any of Apache Pekko Discovery’s lookup mechanisms, specify a service name and lookup timeout in the Couchbase section, and pass in DiscoverySupportDiscoverySupport nodes lookup to enrichAsync and configure Apache Pekko Discovery accordingly.

The Apache Pekko Discovery dependency has to be added explicitly.

Discovery settings (Config discovery)

copysourcepekko.connectors.couchbase {
  session {
    service {
      name = couchbase-service
      lookup-timeout = 1 s
    }
    username = "anotherUser"
    password = "differentPassword"
  }
}

pekko.discovery.method = config
pekko.discovery.config.services = {
  couchbase-service = {
    endpoints = [
      { host = "pekko.apache.org" }
    ]
  }
}

To enable Apache Pekko Discovery on the CouchbaseSessionSettings, use DiscoverySupport.nodes() as enrichment function.

Scala

copysource
val registry = CouchbaseSessionRegistry(actorSystem)

val sessionSettings = CouchbaseSessionSettings(actorSystem)
  .withEnrichAsync(DiscoverySupport.nodes())
val sessionFuture: Future[CouchbaseSession] = registry.sessionFor(sessionSettings, bucketName)

Java

copysourceimport org.apache.pekko.stream.connectors.couchbase.CouchbaseSessionRegistry;
import org.apache.pekko.stream.connectors.couchbase.CouchbaseSessionSettings;
import org.apache.pekko.stream.connectors.couchbase.javadsl.DiscoverySupport;
import org.apache.pekko.stream.connectors.couchbase.javadsl.CouchbaseSession;

CouchbaseSessionRegistry registry = CouchbaseSessionRegistry.get(actorSystem);

CouchbaseSessionSettings sessionSettings =
    CouchbaseSessionSettings.create(actorSystem)
        .withEnrichAsyncCs(DiscoverySupport.getNodes(actorSystem));
CompletionStage<CouchbaseSession> session = registry.getSessionFor(sessionSettings, bucketName);

Reading from Couchbase in Apache Pekko Streams

Using statements

To query Couchbase using the statement DSL use CouchbaseSource.fromStatement.

Scala

copysourceimport com.couchbase.client.java.query.Select.select
import com.couchbase.client.java.query.dsl.Expression._

val resultAsFuture: Future[Seq[JsonObject]] =
  CouchbaseSource
    .fromStatement(sessionSettings, select("*").from(i(queryBucketName)).limit(10), bucketName)
    .runWith(Sink.seq)

Java

copysourceimport static com.couchbase.client.java.query.Select.select;
import static com.couchbase.client.java.query.dsl.Expression.*;

CompletionStage<List<JsonObject>> resultCompletionStage =
    CouchbaseSource.fromStatement(
            sessionSettings, select("*").from(i(queryBucketName)).limit(10), bucketName)
        .runWith(Sink.seq(), actorSystem);

Using N1QL queries

To query Couchbase using the “N1QL” queries use CouchbaseSource.fromN1qlQuery.

Scala

copysourceimport com.couchbase.client.java.query.{ N1qlParams, N1qlQuery }

val params = N1qlParams.build.adhoc(false)
val query = N1qlQuery.simple(s"select count(*) from $queryBucketName", params)

val resultAsFuture: Future[Seq[JsonObject]] =
  CouchbaseSource
    .fromN1qlQuery(sessionSettings, query, bucketName)
    .runWith(Sink.seq)

Java

copysourceimport com.couchbase.client.java.query.N1qlParams;
import com.couchbase.client.java.query.N1qlQuery;

N1qlParams params = N1qlParams.build().adhoc(false);
SimpleN1qlQuery query = N1qlQuery.simple("select count(*) from " + queryBucketName, params);

CompletionStage<JsonObject> resultCompletionStage =
    CouchbaseSource.fromN1qlQuery(sessionSettings, query, bucketName)
        .runWith(Sink.head(), actorSystem);

Get by ID

CouchbaseFlow.fromId methods allow to read documents specified by the document ID in the Apache Pekko Stream.

Scala

copysourceval ids = immutable.Seq("First", "Second", "Third", "Fourth")

val futureResult: Future[immutable.Seq[JsonDocument]] =
  Source(ids)
    .via(
      CouchbaseFlow.fromId(
        sessionSettings,
        bucketName))
    .runWith(Sink.seq)

Java

copysourceList<String> ids = Arrays.asList("First", "Second", "Third", "Fourth");

CompletionStage<List<JsonDocument>> result =
    Source.from(ids)
        .via(CouchbaseFlow.fromId(sessionSettings, queryBucketName))
        .runWith(Sink.seq(), actorSystem);

Writing to Couchbase in Apache Pekko Streams

For each mutation operation we need to create CouchbaseWriteSettingsCouchbaseWriteSettings instance which consists of the following parameters

• Parallelism in access to Couchbase (default 1)
• Couchbase Replication Factor (default ReplicateTo.NONE)
• Couchbase Persistence Level for Write Operation (default PersistTo.NONE)
• 2 seconds operation timeout

These default values are not recommended for production use, as they do not persist to disk for any node.

Scala

copysourceimport com.couchbase.client.java.{ PersistTo, ReplicateTo }
val writeSettings = CouchbaseWriteSettings()
  .withParallelism(3)
  .withPersistTo(PersistTo.FOUR)
  .withReplicateTo(ReplicateTo.THREE)
  .withTimeout(5.seconds)

Java

copysourceCouchbaseWriteSettings writeSettings =
    CouchbaseWriteSettings.create()
        .withParallelism(3)
        .withPersistTo(PersistTo.FOUR)
        .withReplicateTo(ReplicateTo.THREE)
        .withTimeout(Duration.ofSeconds(5));

Read more about durability settings in the Couchbase documentation.

Upsert

The CouchbaseFlow and CouchbaseSink offer factories for upserting documents (insert or update) in Couchbase. upsert is used for the most commonly used JsonDocument, and upsertDoc has as type parameter to support any variants of Document[T]Document<T> which may be RawJsonDocument, StringDocument or BinaryDocument.

The upsert and upsertDoc operators fail the stream on any error when writing to Couchbase. To handle failures in-stream use upsertDocWithResult shown below.

Scala

copysourceval obj = TestObject(id = "First", "First")

val writeSettings = CouchbaseWriteSettings()

val jsonDocumentUpsert: Future[Done] =
  Source
    .single(obj)
    .map(toJsonDocument)
    .via(
      CouchbaseFlow.upsert(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.ignore)

val stringDocumentUpsert: Future[Done] =
  Source
    .single(sampleData)
    .map(toStringDocument)
    .via(
      CouchbaseFlow.upsertDoc(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.ignore)

Java

copysourceCompletionStage<JsonDocument> jsonDocumentUpsert =
    Source.single(obj)
        .map(support::toJsonDocument)
        .via(CouchbaseFlow.upsert(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.head(), actorSystem);

Couchbase writes may fail temporarily for a particular node. If you want to handle such failures without restarting the whole stream, the upsertDocWithResult operator captures failures from Couchbase and emits CouchbaseWriteResult sub-classes CouchbaseWriteSuccess and CouchbaseWriteFailure downstream.

Scala

copysource
val result: Future[immutable.Seq[CouchbaseWriteResult[RawJsonDocument]]] =
  Source(sampleSequence)
    .map(toRawJsonDocument)
    .via(
      CouchbaseFlow.upsertDocWithResult(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.seq)

val failedDocs: immutable.Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
  case res: CouchbaseWriteFailure[RawJsonDocument] => res
}

Java

copysourceimport org.apache.pekko.stream.connectors.couchbase.CouchbaseWriteFailure;
import org.apache.pekko.stream.connectors.couchbase.CouchbaseWriteResult;
CompletionStage<List<CouchbaseWriteResult<StringDocument>>> upsertResults =
    Source.from(sampleSequence)
        .map(support::toStringDocument)
        .via(CouchbaseFlow.upsertDocWithResult(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.seq(), actorSystem);

List<CouchbaseWriteResult<StringDocument>> writeResults =
    upsertResults.toCompletableFuture().get(3, TimeUnit.SECONDS);
List<CouchbaseWriteFailure<StringDocument>> failedDocs =
    writeResults.stream()
        .filter(CouchbaseWriteResult::isFailure)
        .map(res -> (CouchbaseWriteFailure<StringDocument>) res)
        .collect(Collectors.toList());

Replace

The CouchbaseFlow and CouchbaseSink offer factories for replacing documents in Couchbase. replace is used for the most commonly used JsonDocument, and replaceDoc has as type parameter to support any variants of Document[T]Document<T> which may be RawJsonDocument, StringDocument or BinaryDocument.

The replace and replaceDoc operators fail the stream on any error when writing to Couchbase. To handle failures in-stream use replaceDocWithResult shown below.

A replace action will fail if the original Document can’t be found in Couchbase with a DocumentDoesNotExistException.

Scala

copysourceval replaceFuture: Future[Done] =
  Source
    .single(obj)
    .map(toJsonDocument)
    .via(
      CouchbaseFlow.replace(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.ignore)

Java

copysourceimport com.couchbase.client.java.error.DocumentDoesNotExistException;
CompletionStage<JsonDocument> jsonDocumentReplace =
    Source.single(obj)
        .map(support::toJsonDocument)
        .via(CouchbaseFlow.replace(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.head(), actorSystem);
CompletionStage<JsonDocument> jsonDocumentReplace =
    Source.single(obj)
        .map(support::toJsonDocument)
        .via(CouchbaseFlow.replace(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.head(), actorSystem);

Couchbase writes may fail temporarily for a particular node. If you want to handle such failures without restarting the whole stream, the replaceDocWithResult operator captures failures from Couchbase and emits CouchbaseWriteResult sub-classes CouchbaseWriteSuccess and CouchbaseWriteFailure downstream.

Scala

copysource
val result: Future[immutable.Seq[CouchbaseWriteResult[RawJsonDocument]]] =
  Source(sampleSequence)
    .map(toRawJsonDocument)
    .via(
      CouchbaseFlow.upsertDocWithResult(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.seq)

val failedDocs: immutable.Seq[CouchbaseWriteFailure[RawJsonDocument]] = result.futureValue.collect {
  case res: CouchbaseWriteFailure[RawJsonDocument] => res
}

Java

copysourceimport org.apache.pekko.stream.connectors.couchbase.CouchbaseWriteFailure;
import org.apache.pekko.stream.connectors.couchbase.CouchbaseWriteResult;
CompletionStage<List<CouchbaseWriteResult<StringDocument>>> upsertResults =
    Source.from(sampleSequence)
        .map(support::toStringDocument)
        .via(CouchbaseFlow.upsertDocWithResult(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.seq(), actorSystem);

List<CouchbaseWriteResult<StringDocument>> writeResults =
    upsertResults.toCompletableFuture().get(3, TimeUnit.SECONDS);
List<CouchbaseWriteFailure<StringDocument>> failedDocs =
    writeResults.stream()
        .filter(CouchbaseWriteResult::isFailure)
        .map(res -> (CouchbaseWriteFailure<StringDocument>) res)
        .collect(Collectors.toList());

Delete

The CouchbaseFlow and CouchbaseSink offer factories to delete documents in Couchbase by ID.

Scala

copysourceval deleteFuture: Future[Done] =
  Source
    .single(sampleData.id)
    .via(
      CouchbaseFlow.delete(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.ignore)

Java

copysourceCompletionStage<String> result =
    Source.single(sampleData.id())
        .via(CouchbaseFlow.delete(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.head(), actorSystem);

To handle any delete failures such as non-existent documents in-stream, use the the deleteWithResult operator. It captures failures from Couchbase and emits CouchbaseDeleteResults.

Scala

copysourceval deleteResult: Future[CouchbaseDeleteResult] =
  Source
    .single("non-existent")
    .via(
      CouchbaseFlow.deleteWithResult(
        sessionSettings,
        writeSettings,
        bucketName))
    .runWith(Sink.head)

Java

copysourceimport org.apache.pekko.stream.connectors.couchbase.CouchbaseDeleteResult;
CompletionStage<CouchbaseDeleteResult> result =
    Source.single("non-existent")
        .via(CouchbaseFlow.deleteWithResult(sessionSettings, writeSettings, bucketName))
        .runWith(Sink.head(), actorSystem);

Using CouchbaseSession directly

Access via registry

The CouchbaseSesionRegistry is an Apache Pekko extension to manage the life-cycle of Couchbase sessions. All underlying instances are closed upon actor system termination.

When accessing more than one Couchbase cluster, the CouchbaseEnvironment should be shared by setting a single instance for the different CouchbaseSessionSettings.

Scala

copysourceimport com.couchbase.client.java.env.{ CouchbaseEnvironment, DefaultCouchbaseEnvironment }

// Pekko extension (singleton per actor system)
val registry = CouchbaseSessionRegistry(actorSystem)

// If connecting to more than one Couchbase cluster, the environment should be shared
val environment: CouchbaseEnvironment = DefaultCouchbaseEnvironment.create()
actorSystem.registerOnTermination {
  environment.shutdown()
}

val sessionSettings = CouchbaseSessionSettings(actorSystem)
  .withEnvironment(environment)
val sessionFuture: Future[CouchbaseSession] = registry.sessionFor(sessionSettings, bucketName)

Java

copysourceimport com.couchbase.client.java.env.CouchbaseEnvironment;
import com.couchbase.client.java.env.DefaultCouchbaseEnvironment;
import org.apache.pekko.stream.connectors.couchbase.CouchbaseSessionRegistry;
import org.apache.pekko.stream.connectors.couchbase.CouchbaseSessionSettings;
import org.apache.pekko.stream.connectors.couchbase.javadsl.CouchbaseSession;

CouchbaseSessionRegistry registry = CouchbaseSessionRegistry.get(actorSystem);

// If connecting to more than one Couchbase cluster, the environment should be shared
CouchbaseEnvironment environment = DefaultCouchbaseEnvironment.create();
actorSystem.registerOnTermination(() -> environment.shutdown());

CouchbaseSessionSettings sessionSettings =
    CouchbaseSessionSettings.create(actorSystem).withEnvironment(environment);
CompletionStage<CouchbaseSession> sessionCompletionStage =
    registry.getSessionFor(sessionSettings, bucketName);

Manage session life-cycle

Use CouchbaseSessionSettings to get an instance of CouchbaseSession. These settings may be specified in application.conf and complemented in code. Furthermore a session requires the bucket name and needs an ExecutionContext as the creation is asynchronous.

Scala

copysource
implicit val ec: ExecutionContext = actorSystem.dispatcher
val sessionSettings = CouchbaseSessionSettings(actorSystem)
val sessionFuture: Future[CouchbaseSession] = CouchbaseSession(sessionSettings, bucketName)
actorSystem.registerOnTermination(sessionFuture.flatMap(_.close()))

val documentFuture = sessionFuture.flatMap { session =>
  val id = "myId"
  val documentFuture: Future[Option[JsonDocument]] = session.get(id)
  documentFuture.flatMap {
    case Some(jsonDocument) =>
      Future.successful(jsonDocument)
    case None =>
      Future.failed(new RuntimeException(s"document $id wasn't found"))
  }
}

Java

copysourceimport org.apache.pekko.stream.connectors.couchbase.CouchbaseSessionSettings;
import org.apache.pekko.stream.connectors.couchbase.javadsl.CouchbaseSession;

Executor executor = Executors.newSingleThreadExecutor();
CouchbaseSessionSettings sessionSettings = CouchbaseSessionSettings.create(actorSystem);
CompletionStage<CouchbaseSession> sessionCompletionStage =
    CouchbaseSession.create(sessionSettings, bucketName, executor);
actorSystem.registerOnTermination(
    () -> sessionCompletionStage.thenAccept(CouchbaseSession::close));

sessionCompletionStage.thenAccept(
    session -> {
      String id = "myId";
      CompletionStage<Optional<JsonDocument>> documentCompletionStage = session.get(id);
      documentCompletionStage.thenAccept(
          opt -> {
            if (opt.isPresent()) {
              System.out.println(opt.get());
            } else {
              System.out.println("Document " + id + " wasn't found");
            }
          });
    });

Manage bucket life-cycle

For full control a CouchbaseSession may be created from a Couchbase Bucket. See Scalability and Concurrency in the Couchbase documentation for details.

Scala

copysourceimport com.couchbase.client.java.auth.PasswordAuthenticator
import com.couchbase.client.java.{ Bucket, CouchbaseCluster }

val cluster: CouchbaseCluster = CouchbaseCluster.create("localhost")
cluster.authenticate(new PasswordAuthenticator("Administrator", "password"))
val bucket: Bucket = cluster.openBucket("pekko")
val session: CouchbaseSession = CouchbaseSession(bucket)
actorSystem.registerOnTermination {
  session.close()
}

val id = "myId"
val documentFuture = session.get(id).flatMap {
  case Some(jsonDocument) =>
    Future.successful(jsonDocument)
  case None =>
    Future.failed(new RuntimeException(s"document $id wasn't found"))
}

Java

copysourceimport com.couchbase.client.java.Bucket;
import com.couchbase.client.java.CouchbaseCluster;
import com.couchbase.client.java.auth.PasswordAuthenticator;

CouchbaseCluster cluster = CouchbaseCluster.create("localhost");
cluster.authenticate(new PasswordAuthenticator("Administrator", "password"));
Bucket bucket = cluster.openBucket("pekko");
CouchbaseSession session = CouchbaseSession.create(bucket);
actorSystem.registerOnTermination(
    () -> {
      session.close();
      bucket.close();
    });

String id = "First";
CompletionStage<Optional<JsonDocument>> documentCompletionStage = session.get(id);
documentCompletionStage.thenAccept(
    opt -> {
      if (opt.isPresent()) {
        System.out.println(opt.get());
      } else {
        System.out.println("Document " + id + " wasn't found");
      }
    });

To learn about the full range of operations on CouchbaseSession, read the CouchbaseSessionCouchbaseSession API documentation.