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

Author: chedim

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

@@ -4,11 +4,11 @@
 
 package akka.stream.alpakka.couchbase.impl
 
-import akka.{Done, NotUsed}
 import akka.annotation.InternalApi
 import akka.stream.alpakka.couchbase.CouchbaseDocument
 import akka.stream.alpakka.couchbase.scaladsl.{CouchbaseCollectionSession, CouchbaseSession}
 import akka.stream.scaladsl.Source
+import akka.{Done, NotUsed}
 import com.couchbase.client.java.codec.{RawBinaryTranscoder, RawStringTranscoder, Transcoder}
 import com.couchbase.client.java.json.JsonValue
 import com.couchbase.client.java.kv._
@@ -18,8 +18,8 @@ import rx.{Observable, RxReactiveStreams}
 
 import java.util
 import java.util.concurrent.TimeUnit
-import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.{ExecutionContext, Future}
 import scala.jdk.FutureConverters.CompletionStageOps
 
 /**

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

@@ -4,11 +4,11 @@
 
 package akka.stream.alpakka.couchbase.impl
 
-import akka.{Done, NotUsed}
 import akka.annotation.InternalApi
 import akka.stream.alpakka.couchbase.javadsl.CouchbaseSession
 import akka.stream.alpakka.couchbase.{javadsl, scaladsl, CouchbaseDocument}
 import akka.stream.javadsl.Source
+import akka.{Done, NotUsed}
 import com.couchbase.client.java.json.{JsonArray, JsonObject, JsonValue}
 import com.couchbase.client.java.kv.{InsertOptions, RemoveOptions, ReplaceOptions, UpsertOptions}
 import com.couchbase.client.java.manager.query.{CreateQueryIndexOptions, QueryIndex}

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

@@ -4,11 +4,11 @@
 
 package akka.stream.alpakka.couchbase.scaladsl
 
-import akka.{Done, NotUsed}
 import akka.annotation.DoNotInherit
 import akka.stream.alpakka.couchbase.CouchbaseDocument
 import akka.stream.alpakka.couchbase.impl.CouchbaseCollectionSessionImpl
 import akka.stream.scaladsl.Source
+import akka.{Done, NotUsed}
 import com.couchbase.client.java.json.{JsonArray, JsonObject, JsonValue}
 import com.couchbase.client.java.kv.{InsertOptions, RemoveOptions, ReplaceOptions, UpsertOptions}
 import com.couchbase.client.java.manager.query.{CreateQueryIndexOptions, QueryIndex}

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

@@ -13,6 +13,7 @@ import akka.{Done, NotUsed}
 import com.couchbase.client.java._
 import com.couchbase.client.java.json.JsonObject
 import com.couchbase.client.java.query._
+import org.slf4j.LoggerFactory
 
 import java.util.concurrent.atomic.AtomicReference
 import scala.collection.mutable
@@ -24,6 +25,7 @@ import scala.concurrent.{ExecutionContext, Future}
  * @see [[akka.stream.alpakka.couchbase.CouchbaseSessionRegistry]]
  */
 object CouchbaseSession {
+  private val log = LoggerFactory.getLogger(classOf[CouchbaseSession])
 
   /**
    * Create a session against the given bucket. The couchbase client used to connect will be created and then closed when
@@ -68,7 +70,12 @@ object CouchbaseSession {
               enrichedSettings.nodes.mkString(","),
               clusterOptions
             )
-        })
+        }).andThen(c => {
+            log.debug("created couchbase cluster client for " + enrichedSettings.username)
+          }).recover(err => {
+            log.error("failed to create couchbase cluster", err)
+            throw err
+          })
       }
 
 }

couchbase/src/test/java/docs/javadsl/CouchbaseExamplesTest.java

@@ -6,7 +6,6 @@
 
 import akka.Done;
 import akka.actor.ActorSystem;
-import akka.http.scaladsl.server.util.Tuple;
 import akka.stream.Materializer;
 // #deleteWithResult
 import akka.stream.alpakka.couchbase.*;
@@ -16,7 +15,6 @@
 import akka.stream.alpakka.couchbase.javadsl.CouchbaseFlow;
 import akka.stream.alpakka.couchbase.javadsl.CouchbaseSource;
 import akka.stream.alpakka.couchbase.testing.CouchbaseSupportClass;
-import akka.stream.alpakka.couchbase.testing.TestObject;
 import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
@@ -26,28 +24,21 @@
 import com.couchbase.client.java.env.ClusterEnvironment;
 
 import com.couchbase.client.java.json.JsonObject;
-import com.couchbase.client.java.json.JsonValue;
 import org.junit.*;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.*;
+import java.util.stream.Collectors;
 // #registry
 // #session
 import akka.stream.alpakka.couchbase.javadsl.CouchbaseSession;
 // #session
 // #registry
-import java.util.stream.Collectors;
 // #sessionFromBucket
-import com.couchbase.client.java.Bucket;
 // #sessionFromBucket
 
-import scala.Tuple2;
-import scala.concurrent.duration.FiniteDuration;
-
 import static org.hamcrest.CoreMatchers.*;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -221,8 +212,8 @@ public void upsertWithResult() throws Exception {
     List<CouchbaseWriteFailure> failedDocs =
         writeResults.stream()
             .filter(CouchbaseWriteResult::isFailure)
-            .map(res -> (CouchbaseWriteFailure) res)
-            .toList();
+            .map(CouchbaseWriteFailure.class::cast)
+            .collect(Collectors.toUnmodifiableList());
     // #upsertDocWithResult
 
     assertThat(writeResults.size(), is(sampleSequence.size()));
@@ -294,7 +285,7 @@ public void replaceWithResult() throws Exception {
         writeResults.stream()
             .filter(CouchbaseWriteResult::isFailure)
             .map(CouchbaseWriteFailure.class::cast)
-            .toList();
+            .collect(Collectors.toUnmodifiableList());
     // #replaceDocWithResult
 
     assertThat(writeResults.size(), is(list.size()));

couchbase/src/test/resources/discovery.conf

@@ -11,8 +11,8 @@ alpakka.couchbase {
       name = couchbase-service
       lookup-timeout = 1 s
     }
-    username = "anotherUser"
-    password = "differentPassword"
+    username = "Administrator"
+    password = "password"
   }
 }
 

couchbase/src/test/scala/akka/stream/alpakka/couchbase/testing/CouchbaseSupport.scala

@@ -39,7 +39,7 @@ trait CouchbaseSupport {
   val sessionSettings = CouchbaseSessionSettings(actorSystem)
   val bucketName = "akka"
   val scopeName = "alpakka"
-  val collectionName = "karakalpaka"
+  val collectionName = "alpakka-collection"
 
   var session: CouchbaseSession = _
 

couchbase/src/test/scala/docs/scaladsl/CouchbaseFlowSpec.scala

@@ -10,14 +10,14 @@ import akka.stream.alpakka.couchbase.testing.CouchbaseSupport
 import akka.stream.alpakka.couchbase.{CouchbaseDeleteFailure, CouchbaseDeleteResult, CouchbaseDocument}
 import akka.stream.alpakka.testkit.scaladsl.LogCapturing
 import akka.stream.scaladsl.{Sink, Source}
+import akka.stream.testkit.scaladsl.StreamTestKit._
 import com.couchbase.client.core.error.{DocumentNotFoundException, DurabilityImpossibleException}
 import com.couchbase.client.core.msg.kv.DurabilityLevel
+import com.couchbase.client.java.codec.RawBinaryTranscoder
 import com.couchbase.client.java.json.JsonObject
 import com.couchbase.client.java.kv.{RemoveOptions, UpsertOptions}
 import org.scalatest._
 import org.scalatest.concurrent.ScalaFutures
-import akka.stream.testkit.scaladsl.StreamTestKit._
-import com.couchbase.client.java.codec.RawBinaryTranscoder
 
 import scala.collection.immutable
 import scala.concurrent.Future

couchbase/src/test/scala/docs/scaladsl/CouchbaseSessionExamplesSpec.scala

@@ -28,9 +28,15 @@ class CouchbaseSessionExamplesSpec
 
   override implicit def patienceConfig: PatienceConfig = PatienceConfig(10.seconds, 250.millis)
 
-  override def beforeAll(): Unit = super.beforeAll()
-  override def afterAll(): Unit = super.afterAll()
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    upsertSampleData(bucketName, scopeName, collectionName)
+  }
 
+  override def afterAll(): Unit = {
+    cleanAllInCollection(bucketName, scopeName, collectionName)
+    super.afterAll()
+  }
   "a Couchbasesession" should {
     "be managed by the registry" in {
       // #registry

docker-compose.yml

@@ -26,12 +26,12 @@ services:
     ports:
       - "9042:9042"
   couchbase:
-    image: couchbase:community-5.1.1
+    image: couchbase:community-7.6.2
     ports:
       - "8091-8094:8091-8094"
       - "11210:11210"
   couchbase_prep:
-    image: couchbase:community-5.1.1
+    image: couchbase:community-7.6.2
     links:
       - "couchbase"
     entrypoint: ""
@@ -46,6 +46,9 @@ services:
         couchbase-cli bucket-create -c couchbase -u Administrator -p password --bucket akka --bucket-type couchbase --bucket-ramsize 100 --bucket-replica 1 --wait
         couchbase-cli bucket-create -c couchbase -u Administrator -p password --bucket akkaquery --bucket-type couchbase --bucket-ramsize 100 --bucket-replica 1 --wait
         sleep 2 # just wait a tiny bit more after creating the bucket
+        couchbase-cli collection-manage -c couchbase -u Administrator -p password --bucket akka --create-scope alpakka
+        couchbase-cli collection-manage -c couchbase -u Administrator -p password --bucket akka --create-collection alpakka.alpakka-collection
+        sleep 2 # just wait a tiny bit more after creating the collection
         echo 'CREATE PRIMARY INDEX ON akkaquery USING GSI;' | cbq -c Administrator:password -e http://couchbase:8093
       "
   elasticmq: