Refactor ParallelVisitor to support generic exceptions.

Fixes a type-unsafety bug in the declaration of
UnfilteredSkyKeyTTVDTCVisitor.Factory and its use in ParallelSkyQueryUtils.

This is in service of reusing the Parallel visitation framework for traversing
the skyframe graph in TraversalInfoRootPackageExtractor.

RELNOTES:
None.
PiperOrigin-RevId: 262026820
diff --git a/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java b/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
index ac42439..de2a3d1 100644
--- a/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
+++ b/src/main/java/com/google/devtools/build/lib/query2/ParallelSkyQueryUtils.java
@@ -63,7 +63,7 @@
     return env.eval(
         expression,
         context,
-        ParallelVisitor.createParallelVisitorCallback(
+        ParallelVisitorUtils.createParallelVisitorCallback(
             new RdepsUnboundedVisitor.Factory(
                 env, /*unfilteredUniverse=*/ Predicates.alwaysTrue(), callback)));
   }
@@ -77,7 +77,7 @@
     return env.eval(
         expression,
         context,
-        ParallelVisitor.createParallelVisitorCallback(
+        ParallelVisitorUtils.createParallelVisitorCallback(
             new RdepsBoundedVisitor.Factory(
                 env, depth, /*universe=*/ Predicates.alwaysTrue(), callback)));
   }
@@ -91,7 +91,7 @@
     return env.eval(
         expression,
         context,
-        ParallelVisitor.createParallelVisitorCallback(
+        ParallelVisitorUtils.createParallelVisitorCallback(
             new RdepsUnboundedVisitor.Factory(env, unfilteredUniverse, callback)));
   }
 
@@ -111,14 +111,15 @@
                   new ThreadSafeAggregateAllSkyKeysCallback(concurrencyLevel);
               return env.execute(
                   () -> {
-                    Callback<Target> visitorCallback =
-                        ParallelVisitor.createParallelVisitorCallback(
-                            new UnfilteredSkyKeyTTVDTCVisitor.Factory(
+                    UnfilteredSkyKeyTTVDTCVisitor visitor =
+                        new UnfilteredSkyKeyTTVDTCVisitor.Factory(
                                 env,
                                 env.createSkyKeyUniquifier(),
                                 processResultsBatchSize,
-                                aggregateAllCallback));
-                    visitorCallback.process(universeValue);
+                                aggregateAllCallback)
+                            .create();
+                    visitor.visitAndWaitForCompletion(
+                        SkyQueryEnvironment.makeTransitiveTraversalKeysStrict(universeValue));
                     return Predicates.in(aggregateAllCallback.getResult());
                   });
             };
@@ -136,7 +137,7 @@
     return env.eval(
         expression,
         context,
-        ParallelVisitor.createParallelVisitorCallback(
+        ParallelVisitorUtils.createParallelVisitorCallback(
             new RdepsBoundedVisitor.Factory(env, depth, universe, callback)));
   }
 
@@ -165,7 +166,7 @@
     return env.eval(
         expression,
         context,
-        ParallelVisitor.createParallelVisitorCallback(
+        ParallelVisitorUtils.createParallelVisitorCallback(
             new DepsUnboundedVisitor.Factory(env, callback, depsNeedFiltering, context)));
   }