What

• Build infra code to allow for RPC batching in PersisterStateManager.
• Change poll mechanism to be triggered on various events (new RPC, completion of RPC on specific parition key etc.)

Why

Batching

• Batching improves performance by leveraging the turnaround time of an RPC on a specific node n1 (share coord leader) to batch multiple RPCs into a single one for subsequent requests. For example:
  • Suppose we get a Write RPC w1 for n1.
  • We send this RPC in standard way.
  • While w1 is inflight, lets say 5 more RPCs for n1 are received (write rpcs - w2, w3, w4 and read rpcs r1, r2).
  • coalesce write RPCs w2, w3, and w4 into a single RPC W1 and read RPCs r1, r2 as R1.
  • When w1 completes, remove w1 from inflight queue and make 2 RPCs W1 and R1 to n1 and add them to inflight.
  • This saves 3 network calls.
  • The result of W1 and R1 is broken down to create response objects for individual RPCs (w2, w3, w4, r1, r2).
• The batching DS is of type:

Node: {
  RPCType: {
    GroupId: [rpc1, rpc2]
  }
}

• We need RPCType as the key for the inner map as the KafkaApis define request handlers on basis of the APIKey which is the request type. Hence we cannot coalesce a read and write RPC into a single request.
• We need the GroupId as the innermost map key since the RPCs defined in kip-932 can have only one groupId as the parameter. The schemas allow for multiple topicIds and multiple partitions for each topicId.

Polling

• Event based poll is more performant and simpler to implement.

Testing

• Added and updated tests at requisite places.