docs: private <C extends TopologyAffectedCommand & VisitableCommand> InvocationStage retrieveRemoteValues(InvocationContext ctx, C originCommand, List<?> keys) {
        if (trace) {
           log.tracef("Prefetching entries for keys %s using broadcast", keys);
        }
        ClusteredGetAllCommand command = commandsFactory.buildClusteredGetAllCommand(keys, FlagBitSets.SKIP_OWNERSHIP_CHECK, null);
        command.setTopologyId(originCommand.getTopologyId());
        CompletionStage<Map<Address, Response>> rpcFuture = rpcManager.invokeCommandOnAll(command, MapResponseCollector.ignoreLeavers(), rpcManager.getSyncRpcOptions());
        return makeStage(asyncValue(rpcFuture).thenApply(ctx, originCommand, (rCtx, rCommand, rv) -> {
           TopologyAffectedCommand topologyAffectedCommand = (TopologyAffectedCommand) rCommand;
           Map<Address, Response> responseMap = (Map<Address, Response>) rv;
           InternalCacheValue[] maxValues = new InternalCacheValue[keys.size()];
           for (Response response : responseMap.values()) {
              if (!response.isSuccessful()) {
                 throw OutdatedTopologyException.RETRY_NEXT_TOPOLOGY;
              }
              InternalCacheValue[] values = (InternalCacheValue[]) ((SuccessfulResponse) response).getResponseValue();
              int i = 0;
              for (InternalCacheValue icv : values) {
                 if (icv != null) {
                    Metadata metadata = icv.getMetadata();
                    if (metadata instanceof RemoteMetadata) {
                       // not sure if this can happen, but let's be on the safe side
                       throw OutdatedTopologyException.RETRY_NEXT_TOPOLOGY;
                    }
                    if (maxValues[i] == null) {
                       maxValues[i] = icv;
                    } else if (metadata != null && metadata.version() != null) {
                       Metadata maxMetadata;
                       if ((maxMetadata = maxValues[i].getMetadata()) == null || maxMetadata.version() == null
                          || maxMetadata.version().compareTo(metadata.version()) == InequalVersionComparisonResult.BEFORE) {
                          maxValues[i] = icv;
                       }
                    }
                 }
                 ++i;
              }
           }
           Map<Object, InternalCacheValue> map = new HashMap<>(keys.size());
           for (int i = 0; i < maxValues.length; ++i) {
              if (maxValues[i] != null) {
                 map.put(keys.get(i), maxValues[i]);
              }
           }
           if (trace) {
              log.tracef("Prefetched values are %s", map);
           }
           if (map.isEmpty()) {
              return CompletableFutures.completedNull();
           }
           // The put below could fail updating the context if the data container got the updated value while we were
           // prefetching that. Also, we need to call RepeatableReadEntry.updatePreviousValue() to get return value
           // from the main command correct.
           for (Map.Entry<Object, InternalCacheValue> entry : map.entrySet()) {
              entryFactory.wrapExternalEntry(rCtx, entry.getKey(), entry.getValue().toInternalCacheEntry(entry.getKey()), true, true);
           }
           PutMapCommand putMapCommand = commandsFactory.buildPutMapCommand(map, null, STATE_TRANSFER_FLAGS);
           putMapCommand.setTopologyId(topologyAffectedCommand.getTopologyId());
           return invokeNext(rCtx, putMapCommand);
        }));
     }