bulk: Fix handling of symbolioc links

Motivation:
----------

Bulk request keepts target arguments that have been supplied
by the user. These arguments contain target paths that may be
symbolic links. When, during request processing, the resolved
target paths are matched to path keys in arguments map to find
out argument values (like lifetime). If keys contain symbolic
links no match is made and thus user supplied lifetime argument
values are not picked up falling back to default.

Modification:
------------
Commit fff8b79 stores full target
path in target and target arguments structuress. There is no discernable
need to resolve symbolic link as fetch of attributes can be done on
synlinked paths. Remove the step of path resolution in BulkRequestContainerJob
completely.

Result:
-------

Correct behavior of tape rest API when using symbolic links

Issue: dCache#7693
Target: trunk
Request: 10.2, 10.1, 10.0, 9.2
Require-book: no
Require-notes: yes
Signed-off-by: Dmitry Litvintsev <[email protected]>

modules/dcache-bulk/src/main/java/org/dcache/services/bulk/job/BulkRequestContainerJob.java

@@ -120,7 +120,6 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import org.dcache.util.list.ListDirectoryHandler;
 import org.dcache.vehicles.FileAttributes;
 import org.dcache.vehicles.PnfsGetFileAttributes;
-import org.dcache.vehicles.PnfsResolveSymlinksMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -235,7 +234,7 @@ enum ContainerState {
      * proper paths and attributes from listing.
      */
     enum TaskState {
-        RESOLVE_PATH, FETCH_ATTRIBUTES, HANDLE_TARGET, HANDLE_DIR_TARGET
+        FETCH_ATTRIBUTES, HANDLE_TARGET, HANDLE_DIR_TARGET
     }
 
     /**
@@ -464,9 +463,6 @@ void doInner() {
             try {
                 checkForRequestCancellation();
                 switch (state) {
-                    case RESOLVE_PATH:
-                        resolvePath();
-                        break;
                     case FETCH_ATTRIBUTES:
                         fetchAttributes();
                         break;
@@ -512,39 +508,7 @@ void performSync() throws InterruptedException {
         }
 
         /**
-         * (1) symlink resolution on initial targets; bypassed for discovered targets.
-         */
-        private void resolvePath() {
-            LOGGER.debug("{} - resolvePath, resolving {}", ruid, target.getPath());
-            PnfsResolveSymlinksMessage message = new PnfsResolveSymlinksMessage(
-                  target.getPath().toString(), null);
-            ListenableFuture<PnfsResolveSymlinksMessage> requestFuture = pnfsHandler.requestAsync(
-                  message);
-            CellStub.addCallback(requestFuture, new AbstractMessageCallback<>() {
-                @Override
-                public void success(PnfsResolveSymlinksMessage message) {
-                    LOGGER.debug("{} - resolvePath {}, callback success.", ruid, target.getPath());
-                    FsPath path = FsPath.create(message.getResolvedPath());
-                    if (targetPrefix != null && !path.contains(targetPrefix)) {
-                        path = computeFsPath(targetPrefix, path.toString());
-                    }
-                    LOGGER.debug("{} - resolvePath, resolved path {}", ruid, path);
-                    target.setPath(path);
-                    state = TaskState.FETCH_ATTRIBUTES;
-                    taskFuture = executor.submit(TargetTask.this);
-                }
-
-                @Override
-                public void failure(int rc, Object error) {
-                    LOGGER.error("{} - resolvePath, callback failure for {}.", ruid, target);
-                    storeOrUpdate(CacheExceptionFactory.exceptionOf(
-                          rc, Objects.toString(error, null)));
-                }
-            }, callbackExecutor);
-        }
-
-        /**
-         * (2) retrieval of required file attributes.
+         * (1) retrieval of required file attributes.
          */
         private void fetchAttributes() {
             LOGGER.debug("{} - fetchAttributes for path {}", ruid, target.getPath());
@@ -1077,7 +1041,7 @@ private void processFileTargets() {
         for (BulkRequestTarget target : requestTargets) {
             try {
                 checkForRequestCancellation();
-                new TargetTask(target, TaskState.RESOLVE_PATH).submitAsync();
+                new TargetTask(target, TaskState.FETCH_ATTRIBUTES).submitAsync();
             } catch (InterruptedException e) {
                 /*
                  * Cancel most likely called; stop processing.
@@ -1128,4 +1092,4 @@ private void update() {
 
         signalStateChange();
     }
-}
+}