@Override
        public void run() {
            if (inflight == null) {
                throw new IllegalStateException("Bug. Cannot not be handling a client response if there is are no assignments in flight");
            }
            if (responseIsError(response)) {
                requeueAllAfterFailure();
            } else {
                failedAttempts = 0;
                AssignReplicasToDirsResponseData data = ((AssignReplicasToDirsResponse) response.responseBody()).data();

                Set<AssignmentEvent> failed = filterFailures(data, inflight);
                Set<AssignmentEvent> completed = Utils.diff(HashSet::new, new HashSet<>(inflight.values()), failed);
                for (AssignmentEvent assignmentEvent : completed) {
                    if (log.isDebugEnabled()) {
                        log.debug("Successfully propagated assignment {}", assignmentEvent);
                    }
                    assignmentEvent.onComplete();
                }

                if (!failed.isEmpty()) {
                    log.warn("Re-queueing assignments: {}", failed);
                    for (AssignmentEvent event : failed) {
                        pending.put(event.partition, event);
                    }
                }
                inflight = null;
                if (!pending.isEmpty()) {
                    scheduleDispatch();
                }
            }
        }
Handles the response to a dispatched {@link AssignReplicasToDirsRequest}.
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java