@Override
        public void run() {
            log.trace("Received assignment {}", this);
            AssignmentEvent existing = pending.getOrDefault(partition, null);
            boolean existingIsInFlight = false;
            if (existing == null && inflight != null) {
                existing = inflight.getOrDefault(partition, null);
                existingIsInFlight = true;
            }
            if (existing != null) {
                if (existing.dirId.equals(dirId)) {
                    existing.merge(this);
                    log.debug("Ignoring duplicate assignment {}", this);
                    return;
                }
                if (existing.timestampNs > timestampNs) {
                    existing.merge(this);
                    log.debug("Dropping assignment {} because it's older than existing {}", this, existing);
                    return;
                } else if (!existingIsInFlight) {
                    this.merge(existing);
                    log.debug("Dropping existing assignment {} because it's older than {}", existing, this);
                }
            }
            log.debug("Queueing new assignment {}", this);
            pending.put(partition, this);

            if (inflight == null || inflight.isEmpty()) {
                scheduleDispatch();
            }
        }
Handles new generated assignments, to be propagated to the controller.
Assignment events may be handled out of order, so for any two assignment
events for the same topic partition, the one with the oldest timestamp is
disregarded.
server/src/main/java/org/apache/kafka/server/AssignmentsManager.java