es-scroll.scala

val timeout = new TimeValue(600000)

val requestBuilder = client
  .prepareSearch(index)
  .setTypes(indexType)
  .setSize(size.getOrElse(5000))
  .setScroll(timeout)
  .setSearchType(SearchType.SCAN)

query  foreach requestBuilder.setQuery
filter foreach requestBuilder.setFilter
fields map(t => t foreach requestBuilder.addField)
if (fields.exists(_.isEmpty)) requestBuilder.setNoFields()
sort foreach requestBuilder.addSort

log.info("Search query: {}", requestBuilder)

val response = requestBuilder.execute().actionGet()
log.info("Search result: {} total hits", response.getHits.totalHits())

scrollProcess(response, timeout, func)


@annotation.tailrec
final def scrollProcess(scrollResp: SearchResponse, timeout: TimeValue, func: SearchHit ⇒ Unit) {
  var hasError = false
  val scrollRespNext =
    try {
      client.prepareSearchScroll(scrollResp.getScrollId)
        .setScroll(timeout)
        .execute()
        .actionGet()
    } catch {
      case e: Exception ⇒
        log.error("Search error: {}", e)
        Thread.sleep(500)
        hasError = true
        scrollResp // try again
    }

  if (!hasError) {
    scrollRespNext.getHits.hits().foreach(func)
  }

  if (hasError || scrollRespNext.getHits.hits().length != 0) {
    scrollProcess(scrollRespNext, timeout, func)
  }
}