very fast upstream followed by very slow downstream and out of memory issues

Question

I've been taking my experiences gathered from fiddling with akka streams to a practical level and I'm in trouble. Below is my attempt at writing a streaming pipeline that reads specific rows form a very large database table then for each row sends a http call to an external web service than inserts the received responses to another table:

(database library is slick and the http client is play framework's )

val query = (catalogIndices joinLeft catalogSummeries on ((i, s) => i.productId === s.icecatId) filter (t => t._2.isEmpty) map (_._1)).result
val source = Source.fromPublisher(db.stream(query))
source
  .buffer(5, OverflowStrategy.backpressure)
  .mapAsync(5) { i =>
    icecatService.fetchCatalog(i.productId, Languages.en, IcecatService.Contents.all)
  }
  .map { r =>
    for {
      g <- (r.json \ "data" \ "GeneralInfo").validate[GeneralInfo]
      img <- (r.json \ "data" \ "Image").validate[Image]
    } yield (g, img)
  }
  .collect {
    case JsSuccess((g, img), _) => CatalogSummeryData(g.IcecatId, g.Title, g.Brand, g.ProductName, g.BrandPartCode,
      g.GTIN, g.Category.CategoryID, g.Category.Name.Value, img.LowPic, img.ThumbPic)
  }
  .grouped(100)
  .runForeach(s => db.run(catalogSummeries ++= s))

I can tell that the db as the upstream is very fast and those web service calls are really slow. also if I increase the number of parallelism of the mapAsync method much further all my futures timeout. My problem started when I migrated the entire platform of the project from a Vagrant virtual machine to my windows 10. Now every thing in my code runs much much faster but this piece of streaming is always resulting in GC errors or out of memory errors and such. Back in the virtual machine I once left this code running and successfully streamed 4000 results. how should I tweak this code to get it to perform ?

Edit: Also I must mention that on my windows machine absolutely no rows are inserted back before it all shuts down in failure.

EDIT: I receive many types of errors but here's a common one:

[ERROR] [SECURITY][09/27/2017 07:23:33.046] [application-scheduler-1] 
[akka.actor.ActorSystemImpl(application)] Uncaught error from thread 
[application-scheduler-1]: GC overhead limit exceeded, shutting down 
ActorSystem[application]
java.lang.OutOfMemoryError: GC overhead limit exceeded
    at akka.dispatch.AbstractNodeQueue.(AbstractNodeQueue.java:32)
    at akka.actor.LightArrayRevolverScheduler$TaskQueue.

very fast upstream followed by very slow downstream and out of memory issues

0 Answers