Heya. I'm trying to troubleshoot somewhat slow async I/O operator. I need to issue 4 redis calls to perform data enrichment but I'm somewhat unhappy with the performance, I can achieve up to 3K OPS only with this code.

class ItemEnricher(private val redisStoreFactory: RedisStoreFactory) extends RichAsyncFunction[Item, Item] {

  @transient
  implicit private var redisStore: RedisStore = _

  @transient
  implicit private var ec: ExecutionContext = _

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    redisStore = redisStoreFactory.create()
    ec = ExecutionContext.fromExecutor(Executors.directExecutor())
  }

  override def close(): Unit = {
    redisStore.close()

    super.close()
  }

  override def asyncInvoke(input: Item, resultFuture: ResultFuture[Item]): Unit = {
    val userFuture = GetUser.run(input.user_id)
    val catalogFuture = GetCatalog.run(input.catalog_id)
    val promotedItemFuture = GetPromotedItem.run(input.id)
    val itemPhotosFuture = GetItemPhotos.run(input.id)

    val result = for {
      user <- userFuture
      catalog <- catalogFuture
      promotedItem <- promotedItemFuture
      itemPhotos <- itemPhotosFuture
    } yield input.copy(
      item_photo_count = itemPhotos.item_photos.size,
      promoted_until = promotedItem.map(_.promoted_until.getEpochSecond),
      catalog_parent_id = catalog.flatMap(_.parent_id),
      catalog_parent_id_1 = catalog.flatMap(_.parent_id_1),
      catalog_parent_id_2 = catalog.flatMap(_.parent_id_2),
      catalog_parent_id_3 = catalog.flatMap(_.parent_id_3),
      catalog_parent_id_4 = catalog.flatMap(_.parent_id_4),
      user_created_at = user.user.map(_.created_at.getEpochSecond)
    )

    result.onComplete {
      case Success(value) => resultFuture.complete(Seq(value))
      case Failure(exception) => resultFuture.completeExceptionally(exception)
    }
  }
}

Similar jobs that write to redis, perform at 15K OPS. Is there something wrong I'm doing with the execution? Under the hood, I use Lettuce to call redis. I suspect these 4 calls are done sequentially even though I create futures upfront to trigger their parallell execution

How many items are in the

catalog

result? I ask because it looks like the code is iterating over the list 5 times to build the various

catalog_parent_id_x

variables. If this list is large, you could be spending a lot of time iterating over it. An optimization would be to do something like:

var catalog_parent_id = List[T]()
var catalog_parent_id_1 = List[T]()
// other catalogs

catalog.foreach(entry => {
  catalog_parent_id += entry.parent_id
  catalog_parent_id_1 += entry.parent_id_1
  // etc...
})

Hi.

catalog

is a very simple case class:

case class Catalog(
  id: Long,
  title: String,
  parent_id_4: Option[Long] = None,
  parent_id_3: Option[Long] = None,
  parent_id_2: Option[Long] = None,
  parent_id_1: Option[Long] = None,
  parent_id: Option[Long] = None
)

The call from redis returns an `Option[Catalog]`and is using flatMap to assign its

parent_ids

to the

Item

case class, which is briefly this:

case class Item(
  id: Long,
  catalog_parent_id: Option[Long],
  catalog_parent_id_1: Option[Long],
  catalog_parent_id_2: Option[Long],
  catalog_parent_id_3: Option[Long],
  catalog_parent_id_4: Option[Long],
)

I suspect that the issue might actually be single threaded direct executor, but I don't know how to prove it.

Yup. Definitely not an iteration issue if the class is just an Option.

Aye, I've removed some fields from the case class for the sake of brevity, but there are no large collections involved.