handle pekko deprecations (#5555)

Co-authored-by: Brendan Doyle <[email protected]>

Author: bdoyle0182

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala

@@ -174,7 +174,7 @@ object OwSink {
    */
   def combine[T, U, M1, M2](first: Sink[U, M1], second: Sink[U, M2])(
     strategy: Int => Graph[UniformFanOutShape[T, U], NotUsed]): Sink[T, (M1, M2)] = {
-    Sink.fromGraph(GraphDSL.create(first, second)((_, _)) { implicit b => (s1, s2) =>
+    Sink.fromGraph(GraphDSL.createGraph(first, second)((_, _)) { implicit b => (s1, s2) =>
       import GraphDSL.Implicits._
       val d = b.add(strategy(2))
 

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/SplunkLogStore.scala

@@ -33,7 +33,6 @@ import org.apache.pekko.http.scaladsl.model.Uri.Path
 import org.apache.pekko.http.scaladsl.model.headers.Authorization
 import org.apache.pekko.http.scaladsl.model.headers.BasicHttpCredentials
 import org.apache.pekko.http.scaladsl.unmarshalling.Unmarshal
-import org.apache.pekko.stream.OverflowStrategy
 import org.apache.pekko.stream.QueueOfferResult
 import org.apache.pekko.stream.scaladsl.Flow
 import org.apache.pekko.stream.scaladsl.Keep
@@ -178,9 +177,10 @@ class SplunkLogStore(
       .convertTo[String]}: ${l.fields(splunkConfig.logMessageField).convertTo[String].trim}"
 
   //based on https://pekko.apache.org/docs/pekko-http/current/client-side/host-level.html
+  // BoundedSourceQueue automatically drops new elements when full, maintaining dropNew behavior
   val queue =
     Source
-      .queue[(HttpRequest, Promise[HttpResponse])](maxPendingRequests, OverflowStrategy.dropNew)
+      .queue[(HttpRequest, Promise[HttpResponse])](maxPendingRequests)
       .via(httpFlow.getOrElse(defaultHttpFlow))
       .toMat(Sink.foreach({
         case ((Success(resp), p)) => p.success(resp)
@@ -190,7 +190,7 @@ class SplunkLogStore(
 
   def queueRequest(request: HttpRequest): Future[HttpResponse] = {
     val responsePromise = Promise[HttpResponse]()
-    queue.offer(request -> responsePromise).flatMap {
+    queue.offer(request -> responsePromise) match {
       case QueueOfferResult.Enqueued => responsePromise.future
       case QueueOfferResult.Dropped =>
         Future.failed(new RuntimeException("Splunk API Client Queue overflowed. Try again later."))

common/scala/src/main/scala/org/apache/openwhisk/core/database/Batcher.scala

@@ -64,7 +64,7 @@ class Batcher[T, R](batchSize: Int, concurrency: Int, retry: Int)(operation: (Se
       completionMatcher = cm,
       failureMatcher = PartialFunction.empty[Any, Throwable],
       bufferSize = Int.MaxValue,
-      overflowStrategy = OverflowStrategy.dropNew)
+      overflowStrategy = OverflowStrategy.dropBuffer)
     .batch(batchSize, Queue(_))((queue, element) => queue :+ element)
     .mapAsyncUnordered(concurrency) { els =>
       val elements = els.map(_._1)

common/scala/src/main/scala/org/apache/openwhisk/core/database/StoreUtils.scala

@@ -84,7 +84,7 @@ private[database] object StoreUtils {
 
   def combinedSink[T](dest: Sink[ByteString, Future[T]])(
     implicit ec: ExecutionContext): Sink[ByteString, Future[AttachmentUploadResult[T]]] = {
-    Sink.fromGraph(GraphDSL.create(digestSink(), lengthSink(), dest)(combineResult) {
+    Sink.fromGraph(GraphDSL.createGraph(digestSink(), lengthSink(), dest)(combineResult) {
       implicit builder => (dgs, ls, dests) =>
         import GraphDSL.Implicits._
 

common/scala/src/main/scala/org/apache/openwhisk/core/database/azblob/AzureBlobAttachmentStore.scala

@@ -299,7 +299,7 @@ class AzureBlobAttachmentStore(client: BlobContainerAsyncClient, prefix: String,
       case None =>
         blobClient.exists().toFuture.toScala.map { exists =>
           if (exists) {
-            val bbFlux = blobClient.download()
+            val bbFlux = blobClient.downloadStream()
             Some(Source.fromPublisher(bbFlux).map(ByteString.fromByteBuffer))
           } else {
             throw NoDocumentException("Not found on 'readAttachment'.")

common/scala/src/main/scala/org/apache/openwhisk/core/database/s3/S3AttachmentStore.scala

@@ -135,10 +135,14 @@ class S3AttachmentStore(s3Settings: S3Settings, bucket: String, prefix: String,
         s"[ATT_GET] '$prefix' finding attachment '$name' of document 'id: $docId'")
     val source = getAttachmentSource(objectKey(docId, name))
 
-    val f = source.flatMap {
-      case Some(x) => x.withAttributes(s3attributes).runWith(sink)
-      case None    => Future.failed(NoDocumentException("Not found on 'readAttachment'."))
-    }
+    val f = source
+      .flatMap { x =>
+        x.withAttributes(s3attributes).runWith(sink)
+      }
+      .recoverWith {
+        case e: Throwable if isMissingKeyException(e) =>
+          Future.failed(NoDocumentException("Not found on 'readAttachment'."))
+      }
 
     val g = f.transform(
       { s =>
@@ -164,16 +168,14 @@ class S3AttachmentStore(s3Settings: S3Settings, bucket: String, prefix: String,
         s"[ATT_GET] '$prefix' internal error, name: '$name', doc: 'id: $docId', failure: '${failure.getMessage}'")
   }
 
-  private def getAttachmentSource(objectKey: String): Future[Option[Source[ByteString, Any]]] = urlSigner match {
-    case Some(signer) => getUrlContent(signer.getSignedURL(objectKey))
+  private def getAttachmentSource(objectKey: String): Future[Source[ByteString, Any]] = urlSigner match {
+    case Some(signer) =>
+      getUrlContent(signer.getSignedURL(objectKey)).map(_.get)
 
-    // When reading from S3 we get an optional source of ByteString and Metadata if the object exist
-    // For such case drop the metadata
+    // S3.getObject returns Source[ByteString, Future[ObjectMetadata]].
+    // The materialized future will fail if the object doesn't exist, which is handled by the caller.
     case None =>
-      S3.download(bucket, objectKey)
-        .withAttributes(s3attributes)
-        .runWith(Sink.head)
-        .map(x => x.map(_._1))
+      Future.successful(S3.getObject(bucket, objectKey).withAttributes(s3attributes))
   }
 
   private def getUrlContent(uri: Uri): Future[Option[Source[ByteString, Any]]] = {
@@ -182,10 +184,9 @@ class S3AttachmentStore(s3Settings: S3Settings, bucket: String, prefix: String,
       case HttpResponse(status, _, entity, _) if status.isSuccess() && !status.isRedirection() =>
         Future.successful(Some(entity.dataBytes))
       case HttpResponse(status, _, entity, _) =>
-        Unmarshal(entity).to[String].map { err =>
+        Unmarshal(entity).to[String].flatMap { err =>
           //With CloudFront also the error message confirms to same S3 exception format
-          val exp = S3Exception(err, status)
-          if (isMissingKeyException(exp)) None else throw exp
+          Future.failed(S3Exception(err, status))
         }
     }
   }

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Actions.scala

@@ -316,7 +316,7 @@ trait WhiskActionsApi extends WhiskCollectionAPI with PostActionActivation with
         }
       case Failure(t: RecordTooLargeException) =>
         logging.debug(this, s"[POST] action payload was too large")
-        terminate(PayloadTooLarge)
+        terminate(ContentTooLarge)
       case Failure(RejectRequest(code, message)) =>
         logging.debug(this, s"[POST] action rejected with code $code: $message")
         terminate(code, message)

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Entities.scala

@@ -21,7 +21,7 @@ import scala.concurrent.Future
 import scala.language.postfixOps
 import scala.util.Try
 import org.apache.pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
-import org.apache.pekko.http.scaladsl.model.StatusCodes.PayloadTooLarge
+import org.apache.pekko.http.scaladsl.model.StatusCodes.ContentTooLarge
 import org.apache.pekko.http.scaladsl.model.headers.RawHeader
 import org.apache.pekko.http.scaladsl.server.Directive0
 import org.apache.pekko.http.scaladsl.server.Directives
@@ -45,7 +45,7 @@ protected[controller] trait ValidateRequestSize extends Directives {
     new Directive0 {
       override def tapply(f: Unit => Route) = {
         check map {
-          case e: SizeError => terminate(PayloadTooLarge, Messages.entityTooBig(e))
+          case e: SizeError => terminate(ContentTooLarge, Messages.entityTooBig(e))
         } getOrElse f(None)
       }
     }

core/controller/src/main/scala/org/apache/openwhisk/core/controller/Packages.scala

@@ -336,7 +336,7 @@ trait WhiskPackagesApi extends WhiskCollectionAPI with ReferencedEntities {
         logging.debug(this, s"fetching package '$docid' for reference")
         if (docid == wp.docid) {
           logging.error(this, s"unexpected package binding refers to itself: $docid")
-          terminate(UnprocessableEntity, Messages.packageBindingCircularReference(b.fullyQualifiedName.toString))
+          terminate(UnprocessableContent, Messages.packageBindingCircularReference(b.fullyQualifiedName.toString))
         } else {
 
           /** Here's where I check package execute only case with package binding. */

core/controller/src/main/scala/org/apache/openwhisk/core/entitlement/PackageCollection.scala

@@ -102,7 +102,7 @@ class PackageCollection(entityStore: EntityStore)(implicit logging: Logging) ext
             logging.error(this, s"unexpected package binding refers to itself: $doc")
             Future.failed(
               RejectRequest(
-                UnprocessableEntity,
+                UnprocessableContent,
                 Messages.packageBindingCircularReference(binding.fullyQualifiedName.toString)))
           } else {
             checkPackageReadPermission(namespaces, pkgOwner, pkgDocid)