class AbstractCompareAndSwapBasedProxyManager<K> extends AbstractProxyManager<K> {

    private static final CommandResult<?> UNSUCCESSFUL_CAS_RESULT = null;

    protected AbstractCompareAndSwapBasedProxyManager(ClientSideConfig clientSideConfig) {
        super(injectTimeClock(clientSideConfig));
    }

    @Override
    public <T> CommandResult<T> execute(K key, Request<T> request) {
        Timeout timeout = Timeout.of(getClientSideConfig());
        CompareAndSwapOperation operation = timeout.call(requestTimeout -> beginCompareAndSwapOperation(key));
        while (true) {
            CommandResult<T> result = execute(request, operation, timeout);
            if (result != UNSUCCESSFUL_CAS_RESULT) {
                return result;
            }
        }
    }

    @Override
    public <T> CompletableFuture<CommandResult<T>> executeAsync(K key, Request<T> request) {
        Timeout timeout = Timeout.of(getClientSideConfig());
        AsyncCompareAndSwapOperation operation = beginAsyncCompareAndSwapOperation(key);
        CompletableFuture<CommandResult<T>> result = executeAsync(request, operation, timeout);
        return result.thenCompose((CommandResult<T> response) -> retryIfCasWasUnsuccessful(operation, request, response, timeout));
    }

    protected abstract CompareAndSwapOperation beginCompareAndSwapOperation(K key);

    protected abstract AsyncCompareAndSwapOperation beginAsyncCompareAndSwapOperation(K key);

    private <T> CommandResult<T> execute(Request<T> request, CompareAndSwapOperation operation, Timeout timeout) {<FILL_FUNCTION_BODY>}

    private <T> CompletableFuture<CommandResult<T>> retryIfCasWasUnsuccessful(AsyncCompareAndSwapOperation operation, Request<T> request, CommandResult<T> casResponse, Timeout timeout) {
        if (casResponse != UNSUCCESSFUL_CAS_RESULT) {
            return CompletableFuture.completedFuture(casResponse);
        } else {
            return executeAsync(request, operation, timeout).thenCompose((CommandResult<T> response) -> retryIfCasWasUnsuccessful(operation, request, response, timeout));
        }
    }

    private <T> CompletableFuture<CommandResult<T>> executeAsync(Request<T> request, AsyncCompareAndSwapOperation operation, Timeout timeout) {
        return timeout.callAsync(operation::getStateData)
            .thenApply((Optional<byte[]> originalStateBytes) -> originalStateBytes.orElse(null))
            .thenCompose((byte[] originalStateBytes) -> {
                RemoteCommand<T> command = request.getCommand();
                MutableBucketEntry entry = new MutableBucketEntry(originalStateBytes);
                CommandResult<T> result = command.execute(entry, getClientSideTime());
                if (!entry.isStateModified()) {
                    return CompletableFuture.completedFuture(result);
                }

                byte[] newStateBytes = entry.getStateBytes(request.getBackwardCompatibilityVersion());
                return timeout.callAsync(requestTimeout -> operation.compareAndSwap(originalStateBytes, newStateBytes, entry.get(), requestTimeout))
                    .thenApply((casWasSuccessful) -> casWasSuccessful? result : null);
            });
    }

    private static ClientSideConfig injectTimeClock(ClientSideConfig clientSideConfig) {
        if (clientSideConfig.getClientSideClock().isPresent()) {
            return clientSideConfig;
        }
        return clientSideConfig.withClientClock(TimeMeter.SYSTEM_MILLISECONDS);
    }

}

        RemoteCommand<T> command = request.getCommand();
        byte[] originalStateBytes = timeout.call(operation::getStateData).orElse(null);
        MutableBucketEntry entry = new MutableBucketEntry(originalStateBytes);
        CommandResult<T> result = command.execute(entry, getClientSideTime());
        if (!entry.isStateModified()) {
            return result;
        }

        byte[] newStateBytes = entry.getStateBytes(request.getBackwardCompatibilityVersion());
        if (timeout.call(requestTimeout -> operation.compareAndSwap(originalStateBytes, newStateBytes, entry.get(), requestTimeout))) {
            return result;
        } else {
            return null;
        }
    
<methods>public AsyncProxyManager<K> asAsync() ,public RemoteBucketBuilder<K> builder() ,public Optional<io.github.bucket4j.BucketConfiguration> getProxyConfiguration(K) <variables>private static final io.github.bucket4j.distributed.proxy.RecoveryStrategy DEFAULT_RECOVERY_STRATEGY,private static final io.github.bucket4j.distributed.proxy.optimization.Optimization DEFAULT_REQUEST_OPTIMIZER,private AsyncProxyManager<K> asyncView,private final non-sealed io.github.bucket4j.distributed.proxy.ClientSideConfig clientSideConfig