parallel batching of large join requests

[hzding621]

Summary

reduce batch size for really large batched join requests. there are sequential processing in fetchJoin kvstoreFuture.map{...} which becomes the bottleneck when the batch size is really large.

Why / Goal

latency reduction

Test Plan

• Added Unit Tests
• Covered by existing CI
• Integration tested

integration testing:

• observed p99 latency reduction 600ms => 200ms (traffic: batch size ~150, ~17 join parts)

Checklist

• Documentation update

Reviewers

@pengyu-hou @airbnb/airbnb-chronon-maintainers

[nikhil-zlai]

the behavior becomes backwards incompatible - shall we introduce a new method fetchJoinChunked(join, keys, chunkSize) instead?

Also it is better to be explicit with these.

[hzding621]

i'm planning to make it a configurable parameter at Fetcher level.

[hzding621]

@nikhil-zlai why not make the chunking behavior the new default?

[hzding621]

updated to keep existing behavior, ptal

[piyush-zlai]

if we are going to keep this and not split like Nikhil mentioned, it might be good to keep the default behavior as before (unchunked) and allow clients to override if they want - wdyt?

[hzding621]

Good point. updated!

[piyush-zlai]

an idea here for the chunking might be to return individual chunked Futures rather than sequencing. With sequencing you are having to wait for all to get back, with individual futures you are getting a behavior akin to streaming responses back so the caller can start processing / acting on them

[pengyu-hou]

Valid point. This is what GPT said:

val batches = requests.grouped(joinFetchParallelChunkSize.get).toSeq

// Return individual futures instead of a single combined future
val responseFutures: Seq[Future[Seq[Response]]] = batches.map { batch =>
  doFetchJoin(batch, joinConf)
}

// Don't use Future.sequence - instead, process each future individually
responseFutures.foreach { future =>
  future.foreach { responses =>
    // Process each batch of responses as soon as it arrives
    processResponses(responses)  // Your processing function
  }
}

[hzding621]

@piyush-zlai @pengyu-hou thanks for the suggestion! updated and added a separate entry point for a chunked api in both scala and java fetcher

[hzding621]

so I included two flows:

• existing fetchJoin will use Fetcher-level configuration (default to un-chunked). this is useful for clients who'd like to mostly keep existing behavior, but still able to enable chunking (but only at global level). I think Airbnb users will most likely use this flow.
• new fetchJoinChunked for users who'd like to have fine-grained control. this API returns a Seq<Future<Seq<Response>>> for streaming-like behavior.

[pengyu-hou]

should we use something like Option(System.getProperty("ai.chronon.fetcher. join_fetch_parallel_chunk_size")) to pass the default chunk size?