class ManuallySyncingCommandExecutor implements CommandExecutor, AsyncCommandExecutor {

    private static final int ORIGINAL_COMMAND_INDEX = 1;
    private static final int GET_SNAPSHOT_COMMAND_INDEX = 2;

    private final CommandExecutor originalExecutor;
    private final AsyncCommandExecutor originalAsyncExecutor;
    private final OptimizationListener listener;
    private final TimeMeter timeMeter;

    private RemoteBucketState state;

    private long lastSyncTimeNanos;
    private long postponedToConsumeTokens;

    private final ReentrantLock localStateMutationLock = new ReentrantLock();
    private final ReentrantLock remoteExecutionLock = new ReentrantLock();
    private CompletableFuture<?> inProgressSynchronizationFuture = CompletableFuture.completedFuture(null);

    ManuallySyncingCommandExecutor(CommandExecutor originalExecutor, OptimizationListener listener, TimeMeter timeMeter) {
        this.originalExecutor = originalExecutor;
        this.originalAsyncExecutor = null;
        this.listener = listener;
        this.timeMeter = timeMeter;
    }

    ManuallySyncingCommandExecutor(AsyncCommandExecutor originalAsyncExecutor, OptimizationListener listener, TimeMeter timeMeter) {
        this.originalExecutor = null;
        this.originalAsyncExecutor = originalAsyncExecutor;
        this.listener = listener;
        this.timeMeter = timeMeter;
    }

    @Override
    public <T> CommandResult<T> execute(RemoteCommand<T> command) {
        MultiCommand remoteCommand;
        localStateMutationLock.lock();
        try {
            CommandResult<T> localResult = tryConsumeLocally(command);
            if (localResult != null) {
                // remote call is not needed
                listener.incrementSkipCount(1);
                return localResult;
            } else {
                remoteCommand = prepareRemoteCommand(command);
            }
        } finally {
            localStateMutationLock.unlock();
        }

        remoteExecutionLock.lock();
        try {
            CommandResult<MultiResult> remoteResult = originalExecutor.execute(remoteCommand);
            rememberRemoteCommandResult(remoteResult);
            return remoteResult.isError() ?
                (CommandResult<T>) remoteResult :
                (CommandResult<T>) remoteResult.getData().getResults().get(ORIGINAL_COMMAND_INDEX);
        } finally {
            remoteExecutionLock.unlock();
        }
    }

    @Override
    public <T> CompletableFuture<CommandResult<T>> executeAsync(RemoteCommand<T> command) {
        MultiCommand remoteCommand;
        localStateMutationLock.lock();
        try {
            CommandResult<T> result = tryConsumeLocally(command);
            if (result != null) {
                // remote call is not needed
                listener.incrementSkipCount(1);
                return CompletableFuture.completedFuture(result);
            } else {
                remoteCommand = prepareRemoteCommand(command);
            }
        } finally {
            localStateMutationLock.unlock();
        }

        remoteExecutionLock.lock();
        CompletableFuture<CommandResult<MultiResult>> resultFuture;
        try {
            resultFuture = inProgressSynchronizationFuture.thenCompose(f -> originalAsyncExecutor.executeAsync(remoteCommand));
            inProgressSynchronizationFuture = resultFuture.exceptionally((Throwable ex) -> null);
        } finally {
            remoteExecutionLock.unlock();
        }

        return resultFuture.thenApply((CommandResult<MultiResult> remoteResult) -> {
            rememberRemoteCommandResult(remoteResult);
            return remoteResult.isError() ?
                (CommandResult<T>) remoteResult :
                (CommandResult<T>) remoteResult.getData().getResults().get(ORIGINAL_COMMAND_INDEX);
        });
    }

    private <T> CommandResult<T> tryConsumeLocally(RemoteCommand<T> command) {
        long currentTimeNanos = timeMeter.currentTimeNanos();
        if (isNeedToExecuteRemoteImmediately(command, currentTimeNanos)) {
            return null;
        }

        // execute local command
        MutableBucketEntry entry = new MutableBucketEntry(state.copy());
        CommandResult<T> result = command.execute(entry, currentTimeNanos);
        if (result.isConfigurationNeedToBeReplaced()) {
            return null;
        }
        long locallyConsumedTokens = command.getConsumedTokens(result.getData());
        if (locallyConsumedTokens == Long.MAX_VALUE) {
            return null;
        }

        if (!isLocalExecutionResultSatisfiesThreshold(locallyConsumedTokens)) {
            return null;
        }

        postponedToConsumeTokens += locallyConsumedTokens;
        if (entry.isStateModified()) {
            state = entry.get();
        }

        return result;
    }

    private boolean isLocalExecutionResultSatisfiesThreshold(long locallyConsumedTokens) {
        if (locallyConsumedTokens == Long.MAX_VALUE || postponedToConsumeTokens + locallyConsumedTokens < 0) {
            // math overflow
            return false;
        }
        return true;
    }

    private <T> boolean isNeedToExecuteRemoteImmediately(RemoteCommand<T> command, long currentTimeNanos) {
        if (state == null) {
            // was never synchronized before
            return true;
        }

        if (command.isImmediateSyncRequired(postponedToConsumeTokens, currentTimeNanos - lastSyncTimeNanos)) {
            // need to execute immediately because of special command
            return true;
        }

        long commandTokens = command.estimateTokensToConsume();
        if (commandTokens == Long.MAX_VALUE || commandTokens + postponedToConsumeTokens < 0) {
            // math overflow
            return true;
        }

        return false;
    }

    private <T> MultiCommand prepareRemoteCommand(RemoteCommand<T> command) {<FILL_FUNCTION_BODY>}

    private void rememberRemoteCommandResult(CommandResult<MultiResult> remoteResult) {
        localStateMutationLock.lock();
        try {
            lastSyncTimeNanos = timeMeter.currentTimeNanos();
            CommandResult<?> snapshotResult = remoteResult.isError() ? remoteResult : remoteResult.getData().getResults().get(GET_SNAPSHOT_COMMAND_INDEX);
            if (snapshotResult.isError()) {
                state = null;
                return;
            }
            this.state = (RemoteBucketState) snapshotResult.getData();

            // decrease available tokens by amount that consumed while remote request was in progress
            if (postponedToConsumeTokens > 0) {
                this.state.consume(postponedToConsumeTokens);
            }
        } finally {
            localStateMutationLock.unlock();
        }
    }


}

        List<RemoteCommand<?>> commands = new ArrayList<>(3);
        commands.add(new ConsumeIgnoringRateLimitsCommand(this.postponedToConsumeTokens));
        this.postponedToConsumeTokens = 0;
        commands.add(command);
        commands.add(new CreateSnapshotCommand());
        return new MultiCommand(commands);
    
<no_super_class>