class AssignmentLimiter {

    public synchronized void checkAssignFrom(String nodeIdentifier)
    {
        if (offlineNodes.contains(nodeIdentifier)) {
            return;
        }

        long now = ticker.read();
        long start = delayedNodes.computeIfAbsent(nodeIdentifier, key -> now);
        Duration delay = new Duration(now - start, NANOSECONDS);

        if (delay.compareTo(reassignmentDelay) < 0) {
            throw new PrestoException(RAPTOR_REASSIGNMENT_DELAY, format(
                    "Reassignment delay is in effect for node %s (elapsed: %s)",
                    nodeIdentifier,
                    delay.convertToMostSuccinctTimeUnit()));
        }

        if (lastOfflined.isPresent()) {
            delay = new Duration(now - lastOfflined.getAsLong(), NANOSECONDS);
            if (delay.compareTo(reassignmentInterval) < 0) {
                throw new PrestoException(RAPTOR_REASSIGNMENT_THROTTLE, format(
                        "Reassignment throttle is in effect for node %s (elapsed: %s)",
                        nodeIdentifier,
                        delay.convertToMostSuccinctTimeUnit()));
            }
        }

        delayedNodes.remove(nodeIdentifier);
        offlineNodes.add(nodeIdentifier);
        lastOfflined = OptionalLong.of(now);
    }

    @Inject public  AssignmentLimiter(NodeSupplier nodeSupplier, Ticker ticker, MetadataConfig config);
    public  AssignmentLimiter(NodeSupplier nodeSupplier, Ticker ticker, Duration reassignmentDelay, Duration reassignmentInterval);

}