@Override
        public void run() {
            if (inflight != null) {
                throw new IllegalStateException("Bug. Should not be dispatching while there are assignments in flight");
            }
            if (pending.isEmpty()) {
                log.trace("No pending assignments, no-op dispatch");
                return;
            }
            Collection<AssignmentEvent> events = pending.values();
            pending = new HashMap<>();
            inflight = new HashMap<>();
            for (AssignmentEvent event : events) {
                if (inflight.size() < AssignReplicasToDirsRequest.MAX_ASSIGNMENTS_PER_REQUEST) {
                    inflight.put(event.partition, event);
                } else {
                    pending.put(event.partition, event);
                }
            }
            if (!pending.isEmpty()) {
                log.warn("Too many assignments ({}) to fit in one call, sending only {} and queueing the rest",
                        AssignReplicasToDirsRequest.MAX_ASSIGNMENTS_PER_REQUEST + pending.size(),
                        AssignReplicasToDirsRequest.MAX_ASSIGNMENTS_PER_REQUEST);
            }
            Map<TopicIdPartition, Uuid> assignment = inflight.entrySet().stream()
                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().dirId));
            log.debug("Dispatching {} assignments:  {}", assignment.size(), assignment);
            channelManager.sendRequest(new AssignReplicasToDirsRequest.Builder(
                    buildRequestData(brokerId, brokerEpochSupplier.get(), assignment)),
                    new AssignReplicasToDirsRequestCompletionHandler());
        }
Gathers pending assignments and pushes them to the controller in a {@link AssignReplicasToDirsRequest}.
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java