private void backoff(long ms) {
        if (ConnectProtocolCompatibility.fromProtocolVersion(currentProtocolVersion) == EAGER) {
            String stageDescription = "sleeping for a backoff duration of " + ms + "ms";
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                time.sleep(ms);
            }
            return;
        }

        if (backoffRetries > 0) {
            int rebalanceDelayFraction =
                    config.getInt(DistributedConfig.SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG) / 10 / backoffRetries;
            String stageDescription = "sleeping for a backoff duration of " + rebalanceDelayFraction + "ms";
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                time.sleep(rebalanceDelayFraction);
            }
            --backoffRetries;
            return;
        }

        ExtendedAssignment runningAssignmentSnapshot;
        synchronized (this) {
            runningAssignmentSnapshot = ExtendedAssignment.duplicate(runningAssignment);
        }
        log.info("Revoking current running assignment {} because after {} retries the worker "
                + "has not caught up with the latest Connect cluster updates",
                runningAssignmentSnapshot, BACKOFF_RETRIES);
        member.revokeAssignment(runningAssignmentSnapshot);
        backoffRetries = BACKOFF_RETRIES;
    }
Try to read to the end of the config log within the given timeout and capture a fresh snapshot via
{@link ConfigBackingStore#snapshot()}

@param timeoutMs maximum time to wait to sync to the end of the log
@return true if successful; false if timed out
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java