Beautify

Author: pbernet

src/main/scala/alpakka/env/FileServer.scala

@@ -4,7 +4,7 @@ import com.github.blemale.scaffeine.{Cache, Scaffeine}
 import org.apache.pekko.actor.ActorSystem
 import org.apache.pekko.http.scaladsl.Http
 import org.apache.pekko.http.scaladsl.model.StatusCodes.*
-import org.apache.pekko.http.scaladsl.model.{HttpResponse, MediaTypes, StatusCodes}
+import org.apache.pekko.http.scaladsl.model.{HttpResponse, MediaTypes, StatusCode, StatusCodes}
 import org.apache.pekko.http.scaladsl.server.Directives.{logRequestResult, path, *}
 import org.apache.pekko.http.scaladsl.server.{ExceptionHandler, Route}
 import org.slf4j.{Logger, LoggerFactory}
@@ -20,13 +20,14 @@ import scala.util.{Failure, Success}
   *
   * The client can request these types of response:
   *  - HTTP 200 response:        /download/[id]
-  *  - Flaky response:           /downloadflaky/[id]
   *  - Non-idempotent response:  /downloadni/[id]
   *    Allows only one download file request per id, answer with HTTP 404 on subsequent requests
+  *  - Flaky response:           /downloadflaky/[id]
+  *    Reply with additional random failures on requests with certain IDs
   *
   * Uses a cache to remember the "one download per id" behaviour
-  * Note that akka-http also supports server-side caching (by wrapping caffeine in caching directives):
-  * https://doc.akka.io/docs/akka-http/current/routing-dsl/directives/caching-directives/index.html
+  * Note that pekko-http would also support server-side caching (by wrapping caffeine in caching directives):
+  * https://pekko.apache.org/docs/pekko-http/current/common/caching.html
   */
 object FileServer extends App {
   val logger: Logger = LoggerFactory.getLogger(this.getClass)
@@ -116,14 +117,10 @@ object FileServer extends App {
     Thread.sleep(sleepTime.toLong)
   }
 
-  def randomErrorHttpStatusCode = {
-    val statusCodes = Seq(StatusCodes.InternalServerError, StatusCodes.BadRequest, StatusCodes.ServiceUnavailable)
-    val start = 0
-    val end = statusCodes.size - 1
-    val rnd = new scala.util.Random
-    val finalRnd = start + rnd.nextInt((end - start) + 1)
-    val statusCode = statusCodes(finalRnd)
+  def randomErrorHttpStatusCode: StatusCode = {
+    val statusCodes = Seq(StatusCodes.NotFound, StatusCodes.InternalServerError, StatusCodes.ServiceUnavailable)
+    val statusCode = statusCodes(scala.util.Random.nextInt(statusCodes.size))
     logger.info(s" -> Complete with HTTP status code: $statusCode")
-    statusCodes(finalRnd)
+    statusCode
   }
 }

src/main/scala/sample/stream_shared_state/DownloaderRetry.java

@@ -101,21 +101,19 @@ public boolean retryRequest(HttpRequest request, IOException exception, int exec
             } else if (rootCause instanceof SocketException
                     || rootCause instanceof InterruptedIOException
                     || exception instanceof SSLException) {
-                try {
-                    Thread.sleep(DELAY_TO_RETRY_SECONDS * 1000);
-                } catch (InterruptedException e) {
-                    e.printStackTrace(); // do nothing
-                }
+                // Don't sleep here; let getRetryInterval handle the delay
                 return true;
-            } else
+            } else {
                 return false;
+            }
         }
 
         @Override
         public boolean retryRequest(HttpResponse response, int execCount, HttpContext context) {
             int httpStatusCode = response.getCode();
-            if (httpStatusCode != HttpStatus.SC_SERVICE_UNAVAILABLE)
-                return false; // retry only on HTTP 503
+            if (httpStatusCode != HttpStatus.SC_SERVICE_UNAVAILABLE
+                    && httpStatusCode != HttpStatus.SC_INTERNAL_SERVER_ERROR)
+                return false; // retry only on HTTP 500 and 503
 
             if (execCount >= maxRetriesCount) {
                 logger.warn("File downloading failed after {} retries in {} minute(s)",
@@ -130,7 +128,10 @@ public boolean retryRequest(HttpResponse response, int execCount, HttpContext co
 
         @Override
         public TimeValue getRetryInterval(org.apache.hc.core5.http.HttpResponse response, int execCount, HttpContext context) {
-            return TimeValue.ofSeconds(DELAY_TO_RETRY_SECONDS);
+            // Exponential backoff: 2s, 4s, 8s
+            int initialDelaySeconds = 2;
+            long delaySeconds = initialDelaySeconds * (1L << (execCount - 1));
+            return TimeValue.ofSeconds(Math.min(delaySeconds, 10)); // cap at 10 seconds
         }
     }
 

src/main/scala/sample/stream_shared_state/LocalFileCacheCaffeine.scala

@@ -28,7 +28,7 @@ import scala.util.control.NonFatal
   *  - On downstream error: the path needs to be kept longer in the cache
   *  - On restart: populate cache from local filesystem
   *
-  * Before running this class: start [[alpakka.env.FileServer]] to simulate non-idempotent responses
+  * Before running this class: start [[alpakka.env.FileServer]] to simulate non-idempotent (or flaky) responses
   * Monitor `localFileCache` dir with cmd:  watch ls -ltr
   *
   * Doc:
@@ -58,7 +58,7 @@ object LocalFileCacheCaffeine {
   FileUtils.forceMkdir(localFileCache.toFile)
   // Comment out to start with empty local file storage
   // Note that this may provoke "CACHE miss" cases, when we try to scavenge during recoverWith
-  //FileUtils.cleanDirectory(localFileCache.toFile)
+  // FileUtils.cleanDirectory(localFileCache.toFile)
 
 
   def deleteFromFileStore(key: Int, value: Path, cause: caffeine.cache.RemovalCause): Unit = {
@@ -90,6 +90,7 @@ object LocalFileCacheCaffeine {
       .mapAsyncUnordered(5) { message =>
         def processNext(message: Message): Message = {
           val key = message.id
+          // switch to /downloadflaky for "more confusion"
           val url = new URI("http://127.0.0.1:6001/downloadni/" + key.toString)
           val destinationFile = localFileCache.resolve(Paths.get(message.id.toString + ".zip"))