Shekharrajak commented on code in PR #22835:
URL: https://github.com/apache/kafka/pull/22835#discussion_r3695449693
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -179,10 +181,59 @@ Map<Node, Map<RPCType, Map<String,
List<PersisterStateManagerHandler>>>> nodeRPC
return nodeRPCMap;
}
+ // test visibility
+ Collection<RequestAndCompletionHandler> generateRequests() {
+ return sender.generateRequests();
+ }
+
public void setGenerateCallback(Runnable generateCallback) {
this.generateCallback = generateCallback;
}
+ /**
+ * Demultiplexes a combined RPC response into topicId -> partition ->
per-partition result, so that
+ * each handler of a coalesced request finds its own slice in O(1) rather
than rescanning the whole
+ * response (which costs O(N) per handler, hence O(N^2) per batch).
+ *
+ * @param <P> per-partition result type of the RPC, e.g. {@link
WriteShareGroupStateResponseData.PartitionResult}
+ */
+ static final class PartitionResultIndex<P> {
+ private final Map<Uuid, Map<Integer, P>> index;
+
+ private PartitionResultIndex(Map<Uuid, Map<Integer, P>> index) {
+ this.index = index;
+ }
+
+ /**
+ * @param results per-topic results of the combined response
+ * @param topicId extracts the topic id from a per-topic result
+ * @param partitions extracts the per-partition results of a per-topic
result
+ * @param partition extracts the partition number from a
per-partition result
+ * @param <T> per-topic result type of the RPC, e.g. {@link
WriteShareGroupStateResponseData.WriteStateResult}
+ * @param <P> per-partition result type of the RPC
+ */
+ static <T, P> PartitionResultIndex<P> build(
+ List<T> results,
+ Function<T, Uuid> topicId,
+ Function<T, List<P>> partitions,
+ ToIntFunction<P> partition
+ ) {
+ Map<Uuid, Map<Integer, P>> index = new HashMap<>();
+ for (T result : results) {
+ Map<Integer, P> byPartition =
index.computeIfAbsent(topicId.apply(result), t -> new HashMap<>());
+ for (P p : partitions.apply(result)) {
+ byPartition.put(partition.applyAsInt(p), p);
Review Comment:
map inserts once and O(1) lookup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
