Couchbase: Updates the integration to support collections #3396

chedim · 2025-05-08T15:22:24Z

References #3395

This PR introduces new CouchbaseCollectionSession class that can be accessed from CouchbaseSession via CouchbaseSession::collection method. It also updates sinks, sources and flow constructors and their signatures to work with collections.

All of these are breaking changes.

Please note that, although fascinated by the framework and some language features, I am not very familiar with either alpakka or scala and more of a Java dev. So, any suggestions on how to improve my changes are greatly appreciated.

Cheers!

…se dependencies; Tests are passing but I need to add more coverage.

lightbend-cla-validator · 2025-05-08T15:22:29Z

Hi @chedim,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Akka Contributors License Agreement:

https://www.lightbend.com/contribute/cla

chedim · 2025-05-08T15:31:25Z

Hi @chedim,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Akka Contributors License Agreement:

https://www.lightbend.com/contribute/cla

Working on it. Hopefully, it'll not take long 🤞

...hbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala

lightbend-cla-validator · 2025-05-19T15:57:59Z

Hi @chedim,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Akka Contributors License Agreement:

https://www.lightbend.com/contribute/cla

ennru · 2025-06-02T15:32:19Z

We have received the CLA.

chedim · 2025-06-03T16:19:30Z

We signed the CLA. Authorized logins: chedim and deniswrosa

sebastian-alfers · 2025-06-05T07:42:16Z

Closing and re-opening to trigger the CLA validator again.

…comment.

chedim · 2025-07-10T16:16:35Z

@ennru @sebastian-alfers @leviramsey Could somebody please review the PR whenever they get a chance? Thank you!

patriknw

A few initial comments/questions...

...hbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseFlow.scala

ennru · 2025-09-30T12:26:05Z

@chedim We will likely do an Alpakka release in the near future, is this something you would be able to get back to?

…ents

chedim · 2025-09-30T22:57:36Z

@ennru , Updated the PR according to comments — improved method signatures, replaced tuples with CouchbaseDocument and put futureFlows as requested. Please re-review and LMK if any additional changes are needed.

Thank you!

ennru · 2025-10-02T13:26:25Z

I pushed a few formalities forward, but there are even compilation errors.

https://github.com/akka/alpakka/actions/runs/18194021044/job/51795533162?pr=3396#step:6:494

The docs still try to include snippets that don't exist in your new implementation

https://github.com/akka/alpakka/actions/runs/18194021044/job/51795533162?pr=3396#step:5:642

The last obvious bit is MiMa, which triggers as the API changed. That is OK and we can add filters to silence it.

chedim · 2025-10-06T16:00:28Z

Looking into the compilation issues, sorry for missing them.

- Fixes a compilation error under jdk11; - updates unit tests to work with docker-compose.yaml docker couchbase instance; - updates couchbase_prep compose service to create required collection - updates to documentation related to the new APIs

chedim · 2025-10-10T00:05:18Z

HI @ennru, I've updated the PR — compilation errors were on jdk11 (and I was running tests with jdk24, silly me). I've also updated the documentation and docker-compose definitions for ci so that they use newer CB images and create required for tests collections.

Tested under jdk11 with docker compose up -d couchbase_prep && sbt couchbase/test and sbt docs/previewSite (under jdk24, should I test it under 11?)

please lmk if there are any suggestions or changes you'd like me to make. Thank you!

chedim · 2025-10-21T16:22:30Z

Updated the PR with sbt scalafmtAll and a fix for the JSON test failure.

patriknw

looking good, only a few minor things

couchbase/src/main/scala/akka/stream/alpakka/couchbase/CouchbaseDocument.scala

patriknw · 2025-10-22T11:43:14Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/CouchbaseSessionRegistry.scala

        .clusterFor(key.settings)
-        .flatMap(cluster => CouchbaseSession(cluster, key.bucketName)(blockingDispatcher))(
+        .flatMap(cluster => CouchbaseSession(cluster, key.bucketName))(
          ExecutionContext.parasitic


It's important to only use ExecutionContext.parasitic when it's absolutely certain that there will be no exceptions. That might not be the case here? createClusterClient?

I've returned the code that was using blockingDispatcher, please lmk if you want to use something else here.
Thank you!

patriknw · 2025-10-22T11:53:05Z

...hbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseCollectionSession.scala

+   * @tparam T type of the object to return
+   * @return a future that completes with created object or errors if the operation failed
+   */
+  def get[T](id: String, target: Class[T]): Future[CouchbaseDocument[T]]


In Scala it would be more idiomatic to use:

def get[T: ClassTag](id: String): Future[CouchbaseDocument[T]]

If you need the actual class of that ClassTag you can use:

val clazz = implicitly[ClassTag[T]].runtimeClass

I went ahead and applied this suggestion everywhere I was using generics, please lmk if that's not what you meant.

Perhaps that was going too far. I haven't looked at all usages, but in general it should only be used when you actually need the Class or don't have the T as another parameter. This get[T] is a good example. I wouldn't use it as a general replacement for [T].

Also, it must not leak to any javadsl, since this is a Scala only thing.

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseFlow.scala

chedim · 2025-10-27T23:24:17Z

@patriknw , updated the PR according to the comments, had couple of questions if I understood the comments correctly.
Thank you for the review!

ennru · 2025-10-28T16:40:02Z

A compiler warning in the example needs to be avoided:

[warn] /home/runner/work/alpakka/alpakka/couchbase/src/test/java/docs/javadsl/CouchbaseExamplesTest.java:206:41: found raw type: akka.stream.alpakka.couchbase.CouchbaseDocument
[warn]   missing type arguments for generic class akka.stream.alpakka.couchbase.CouchbaseDocument<T>
[warn] CouchbaseDocument
[warn]                  ^
[warn] /home/runner/work/alpakka/alpakka/couchbase/src/test/java/docs/javadsl/CouchbaseExamplesTest.java:206:37: unchecked call to CouchbaseDocument(java.lang.String,T,scala.reflect.ClassTag<T>) as a member of the raw type akka.stream.alpakka.couchbase.CouchbaseDocument
[warn] new CouchbaseDocument("First", "First", ClassTag.apply(String.class))
[warn]                                     ^
[error] /home/runner/work/alpakka/alpakka/couchbase/src/test/java/docs/javadsl/CouchbaseExamplesTest.java: warnings found and -Werror specified

chedim · 2025-10-29T02:42:50Z

Should be fixed now.

johanandren

Did a round of reviewing with some feedback

johanandren · 2025-10-30T10:57:09Z

...hbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala

+
+  override def insert[T: ClassTag](id: String, document: T, insertOptions: InsertOptions): Future[Done] = {
+    if (insertOptions.build.transcoder() == null) {
+      insertOptions.transcoder(chooseTranscoder[T]())


Not great mutating a parameter passed in through a public API like this but since InsertOptions is mutable and there is no copy method in that API I guess there is now way around it.

Thank you for pointing this out. I think it is better to remove this altogether and have the client error out if it is unable to select a transcoder or if it was not provided — the developers can always fix that by adding an explicit transcoder into the options.

johanandren · 2025-10-30T11:01:52Z

...hbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala

+    val target: Class[T] = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
+    underlying
+      .get(id, GetOptions.getOptions.transcoder(chooseTranscoder[T]()))
+      .thenApply(gr => new CouchbaseDocument[T](id, gr.contentAs(target)))


This will likely run on the common FJP if not already completed when thenApply is invoked (in which case it executes on the calling thread), better do it as a map on the Scala Future where execution context is explicit, to keep it on the same thread pool as the other stuff here.

Same for the other thenApplys in this class.

johanandren · 2025-10-30T11:06:22Z

...rc/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionJavaAdapter.scala

+                                    document: T,
+                                    replaceOptions: ReplaceOptions,
+                                    timeout: Duration): CompletionStage[Done] =
+    delegate.replace(id, document, replaceOptions, FiniteDuration.apply(timeout.toNanos, TimeUnit.NANOSECONDS)).asJava


use .toScala for the Duration instead, from import scala.jdk.DurationConverters._

johanandren · 2025-10-30T11:10:19Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseSessionImpl.scala

+  override def underlying: AsyncBucket = cluster.bucket(bucketName)
+
+  override def streamedQuery(query: String): Source[JsonObject, NotUsed] = {
+    Source.fromIterator(


This was an async stream, but now switched to a blocking iterator, is there no way to get an asynchronous stream/publisher from the client anymore?

I'm updating it to use RxReactiveStreams as it was before. Unfortunately, there is no way to stream results from the server as they arrive, only after all of them are received and collected into a list.

johanandren · 2025-10-30T11:12:29Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseSessionImpl.scala

+  override def singleResponseQuery(query: String): Future[Option[JsonObject]] =
+    cluster
+      .query(query)
+      .thenApply(getSingleResult(_))


Same as abovve with thenApply here

johanandren · 2025-10-30T11:37:03Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSession.scala

+              enrichedSettings.nodes.mkString(","),
+              clusterOptions
+            )
+        }).andThen(c => {


Might be intentional, but just to make sure c in andThen is a Try and this will execute regardless if it failed or succeeded.

Also a few style things: this is Java syntax for lambdas, in Scala we'd generally use

.andThen { c => ... ... }

or

.andThen({ c => ... ... })

When a parameter is not actually used, use _ to mark it as ignored.

johanandren · 2025-10-30T11:38:27Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSession.scala

+              clusterOptions
+            )
+        }).andThen(c => {
+            log.debug("created couchbase cluster client for " + enrichedSettings.username)


Use placeholders for logging:

Suggested change

log.debug("created couchbase cluster client for " + enrichedSettings.username)

log.debug("created couchbase cluster client for {}", enrichedSettings.username)

This avoids the string concatenation in case debug logging is disabled.

johanandren · 2025-10-30T11:42:30Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSession.scala

+          .recover(err => {
+            log.error("failed to create couchbase cluster", err)
+            throw err
+          })


Since the only intent is to side effect (the logging) this could potentially be covered by a singe andThen:

.andThen { result => log.debug("created couchbase cluster client for {}", enrichedSettings.username) result.failed.foreach { exception => log.error("failed to create couchbase cluster", exception) } }

We are already returning a failed future with the exception though, so logging at error here seems superflous?

johanandren · 2025-10-30T11:53:20Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/scaladsl/CouchbaseSession.scala

-  def counter(id: String, delta: Long, initial: Long, writeSettings: CouchbaseWriteSettings): Future[Long]
+  private val collectionSessions = new AtomicReference(
+    mutable.WeakHashMap.empty[(String, String), CouchbaseCollectionSession]
+  )


The general pattern for doing this kind of compare and swap is that the collection inside would be immutable.

I see this is safe since it is probably safe since cloned before adding an entry on each try, but better to use Java concurrent hash map in this case.

WeakHashMap and weak references are problematic for GC in general so I'd recommend not to try to use that as a cache eviction strategy.

If a pool is really needed it'd be better be done as an akka extension, but I think we should take a step back and consider if this is really worth it. And if it is worth a pool of clients, let's do that a s a separate PR.

My thinking here was very similar — the number of collection sessions probably would be low and they will be reused often so, this shouldn't result in GC issues and having a pool of sessions feels a little bit as overkill here.

The problematic aspect, as I remember it is not the number of weak references, but using weak references at all, messing with GC. Better push that choice to user space and not do it in the library.

Static singleton caches like this also has problems for testing and if multiple actor systems are used in the same JVM for some reason, so we generally avoid that anyway.

johanandren · 2025-10-30T11:55:10Z

couchbase/src/main/scala/akka/stream/alpakka/couchbase/CouchbaseDocument.scala

+
+import scala.reflect.ClassTag
+
+class CouchbaseDocument[T: ClassTag](val id: String, val document: T) {


Is the constructor part of the public API also for Java? (That could make sense for example for users constructing CouchBaseDocument instances in their tests) In that case T : ClassTag is problematic and there should be an explicit constructor for Java accepting Class<T>.

Also, make it final if not intended for user subclassing as public API.

Updates Couchbase integration to support collections; Updates Couchba…

a938d56

…se dependencies; Tests are passing but I need to add more coverage.

probot-autolabeler bot added dependency-change For PRs changing the version of a dependency. p:couchbase labels May 8, 2025

chedim commented May 8, 2025

View reviewed changes

...hbase/src/main/scala/akka/stream/alpakka/couchbase/impl/CouchbaseCollectionSessionImpl.scala Outdated Show resolved Hide resolved

Brings back concurrency settings, updates documentation comments

9e40ece

sebastian-alfers closed this Jun 5, 2025

sebastian-alfers reopened this Jun 5, 2025

chedim marked this pull request as ready for review June 30, 2025 14:39

chedim requested a review from leviramsey July 1, 2025 16:45

Fixes issues with transcoder selection and optimizes the code per PR …

824b61b

…comment.

patriknw reviewed Aug 19, 2025

View reviewed changes

Updates the integration to not use tuples and addresses other PR comm…

982701c

…ents

chedim force-pushed the 3395-update-couchbase branch from 99e1633 to 982701c Compare September 30, 2025 22:56

chedim requested a review from patriknw October 1, 2025 15:08

ennru added 2 commits October 2, 2025 14:33

formalities

8ab5baf

scalafmt

228f14b

chedim changed the title ~~Updates Couchbase integration to support collections~~ Couchbase: Updates the integration to support collections Oct 9, 2025

Updates to couchbase integration:

55c35fc

- Fixes a compilation error under jdk11; - updates unit tests to work with docker-compose.yaml docker couchbase instance; - updates couchbase_prep compose service to create required collection - updates to documentation related to the new APIs

chedim force-pushed the 3395-update-couchbase branch from 884c976 to 55c35fc Compare October 9, 2025 23:38

probot-autolabeler bot added the documentation label Oct 9, 2025

scalafmt and fix for JSON tests

8ddfda8

patriknw reviewed Oct 22, 2025

View reviewed changes

Addresses PR comments for Couchbase integration

ec0bbed

chedim force-pushed the 3395-update-couchbase branch from 154ceab to ec0bbed Compare October 27, 2025 23:14

Fixes a missed diamond in CouchbaseExamplesTest

a1c239e

mima filter for everything couchbase

9d75f8d

johanandren reviewed Oct 30, 2025

View reviewed changes

	log.debug("created couchbase cluster client for " + enrichedSettings.username)
	log.debug("created couchbase cluster client for {}", enrichedSettings.username)


		import scala.reflect.ClassTag

		class CouchbaseDocument[T: ClassTag](val id: String, val document: T) {

Couchbase: Updates the integration to support collections #3396

Are you sure you want to change the base?

Couchbase: Updates the integration to support collections #3396

Uh oh!

Conversation

chedim commented May 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lightbend-cla-validator commented May 8, 2025

Uh oh!

chedim commented May 8, 2025

Uh oh!

Uh oh!

lightbend-cla-validator commented May 19, 2025

Uh oh!

ennru commented Jun 2, 2025

Uh oh!

chedim commented Jun 3, 2025

Uh oh!

sebastian-alfers commented Jun 5, 2025

Uh oh!

chedim commented Jul 10, 2025

Uh oh!

patriknw left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ennru commented Sep 30, 2025

Uh oh!

chedim commented Sep 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ennru commented Oct 2, 2025

Uh oh!

chedim commented Oct 6, 2025

Uh oh!

chedim commented Oct 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chedim commented Oct 21, 2025

Uh oh!

patriknw left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

chedim commented Oct 27, 2025

Uh oh!

ennru commented Oct 28, 2025

Uh oh!

chedim commented Oct 29, 2025

Uh oh!

johanandren left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

chedim commented May 8, 2025 •

edited

Loading

chedim commented Sep 30, 2025 •

edited

Loading

chedim commented Oct 10, 2025 •

edited

Loading