docs: final KeyAwareLockPromise lockAllOrRegisterBackupLock(TxInvocationContext<?> ctx, Collection<?> keys,
                                                                    long lockTimeout) throws InterruptedException {
        if (keys.isEmpty()) {
           return KeyAwareLockPromise.NO_OP;
        }
  
        final Log log = getLog();
        Collection<Object> keysToLock = new ArrayList<>(keys.size());
  
        LocalizedCacheTopology cacheTopology = cdl.getCacheTopology();
        for (Object key : keys) {
           switch (cacheTopology.getDistribution(key).writeOwnership()) {
              case PRIMARY:
                 if (trace) {
                    log.tracef("Acquiring locks on %s.", toStr(key));
                 }
                 keysToLock.add(key);
                 break;
              case BACKUP:
                 if (trace) {
                    log.tracef("Acquiring backup locks on %s.", toStr(key));
                 }
                 ctx.getCacheTransaction().addBackupLockForKey(key);
                 break;
              default:
                 break;
           }
        }
  
        if (keysToLock.isEmpty()) {
           return KeyAwareLockPromise.NO_OP;
        }
  
        return checkPendingAndLockAllKeys(ctx, keysToLock, lockTimeout);
     }