dcache-bulk: mark expanded directory skipped if activity does not app…

…ly to directories

Motivation:

https://rb.dcache.org/r/13782/
master@771db06b8a52b18939a87171a480c9d7776814fc

changed behavior to store targets immediately.
This change, however, means that directories
included among the initial paths are now
stored as targets, whether they are to be
acted upon (other than by being expanded)
––for instance by deletion––or not.  In the
case that the activity (such as PIN) does
not apply to these directories, they
need to be marked as SKIPPED.  Otherwise,
the request completion logic (counting
targets in final states) will prevent
the request from completing.

Modification:

Add the check for the expanded directory
and store it as SKIPPED.

Result:

Requests terminate as they should.

Target: master
Request: 8.2
Requires-notes: yes
Patch: https://rb.dcache.org/r/13818/
Acked-by: Tigran

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

@@ -64,6 +64,7 @@ LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.COMPLETED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.FAILED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.State.READY;
+import static org.dcache.services.bulk.util.BulkRequestTarget.State.SKIPPED;
 import static org.dcache.services.bulk.util.BulkRequestTarget.computeFsPath;
 
 import com.google.common.collect.Range;
@@ -369,7 +370,7 @@ protected DirectoryStream getDirectoryListing(FsPath path)
     }
 
     protected void expandDepthFirst(FsPath path, FileAttributes dirAttributes)
-          throws CacheException, InterruptedException {
+          throws CacheException, InterruptedException, BulkStorageException {
         checkForRequestCancellation();
 
         DirectoryStream stream = getDirectoryListing(path);
@@ -414,6 +415,16 @@ protected void expandDepthFirst(FsPath path, FileAttributes dirAttributes)
             case BOTH:
             case DIR:
                 handleDirTarget(path, dirAttributes);
+                break;
+            case FILE:
+                /*
+                 * Because we now store all initial targets immediately,
+                 * we need to mark such a directory as SKIPPED; otherwise
+                 * the request will not complete on the basis of querying for
+                 * completed targets and finding this one unhandled.
+                 */
+                targetStore.storeOrUpdate(toTarget(path, Optional.of(dirAttributes),
+                          SKIPPED, null));
         }
     }
 

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

@@ -355,7 +355,7 @@ private void storeAll(List<String> targets) throws InterruptedException {
                 } else if (info.attributes.getFileType() != FileType.SPECIAL) {
                     store(info.path, info.attributes);
                 }
-            } catch (CacheException e) {
+            } catch (BulkServiceException | CacheException e) {
                 LOGGER.error("storeAll {}, path {}, error {}.", rid, info.path, e.getMessage());
             }
         }

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

@@ -202,7 +202,7 @@ private void handleTarget(FsPath path) throws InterruptedException {
             } else if (attributes.getFileType() != FileType.SPECIAL) {
                 perform(path, attributes);
             }
-        } catch (CacheException e) {
+        } catch (BulkServiceException | CacheException e) {
             LOGGER.error("handleTarget {}, path {}, error {}.", rid, path, e.getMessage());
             register(path, Futures.immediateFailedFuture(e), attributes, e);
         }