Merge pull request #315 from Acxiom/develop

Release 1.8.7
Acxiom · Jul 6, 2022 · 2d56553 · 2d56553
2 parents e29e5e2 + 784badb
commit 2d56553
Show file tree

Hide file tree

Showing 12 changed files with 19 additions and 17 deletions.
diff --git a/metalus-application/pom.xml b/metalus-application/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
     <dependencies>
         <dependency>

diff --git a/metalus-aws/pom.xml b/metalus-aws/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/metalus-common/pom.xml b/metalus-common/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/metalus-core/pom.xml b/metalus-core/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/metalus-core/src/main/scala/com/acxiom/pipeline/PipelineDependencyExecutor.scala b/metalus-core/src/main/scala/com/acxiom/pipeline/PipelineDependencyExecutor.scala
@@ -146,13 +146,14 @@ object PipelineDependencyExecutor {
               DependencyResult(execution, None,
                 Some(new IllegalArgumentException(s"fork execution ${execution.id} forkByValues ${execution.forkByValue.get} does not point to a valid list")))
             } else {
-              val forkedProcesses = forkValues.asInstanceOf[List[Any]].map(forkValue => {
+              val forkedProcesses = forkValues.asInstanceOf[List[Any]].zipWithIndex.map{ case (forkValue, forkIndex) =>
                 val ctx = execution.pipelineContext.setGlobal("executionForkValue", forkValue)
+                  .setGlobal("executionForkValueIndex", forkIndex)
                 Future {
                   processFutures(List(startExecution(execution.asInstanceOf[DefaultPipelineExecution].copy(pipelineContext = ctx))),
                     Map[String, DependencyResult](), executionGraph)
                 }
-              })
+              }
               // Wait for the forks to complete
               Await.ready(Future.sequence(forkedProcesses), Duration.Inf)
               val finalFutureMap = forkedProcesses.foldLeft(FutureMap(List(), Map[String, DependencyResult]()))((futureMap, f) => {
@@ -169,10 +170,11 @@ object PipelineDependencyExecutor {
               })
               // Execute the join
               val joinExecution = findJoinExecution(execution, executionGraph, execution).get
-              DependencyResult(PipelineExecution(joinExecution.id, joinExecution.pipelines, joinExecution.initialPipelineId,
-                prepareExecutionContext(joinExecution, finalFutureMap.resultMap), joinExecution.parents, None, None, "join"),
-                Some(PipelineExecutor.executePipelines(execution.pipelines, execution.initialPipelineId,
-                  execution.pipelineContext.setGlobal("executionId", execution.id))), None, Some(finalFutureMap))
+              val joinWithResults = PipelineExecution(joinExecution.id, joinExecution.pipelines, joinExecution.initialPipelineId,
+                prepareExecutionContext(joinExecution, finalFutureMap.resultMap), joinExecution.parents, None, None, "join")
+              DependencyResult(joinWithResults,
+                Some(PipelineExecutor.executePipelines(joinWithResults.pipelines, joinWithResults.initialPipelineId,
+                  joinWithResults.pipelineContext.setGlobal("executionId", joinWithResults.id))), None, Some(finalFutureMap))
             }
           } catch {
             case t: Throwable => DependencyResult(execution, None, Some(t))

diff --git a/metalus-delta/pom.xml b/metalus-delta/pom.xml
@@ -12,7 +12,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/metalus-examples/pom.xml b/metalus-examples/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/metalus-gcp/pom.xml b/metalus-gcp/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <dependencyManagement>

diff --git a/metalus-kafka/pom.xml b/metalus-kafka/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/metalus-mongo/pom.xml b/metalus-mongo/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/metalus-utils/pom.xml b/metalus-utils/pom.xml
@@ -9,7 +9,7 @@
     <parent>
         <groupId>com.acxiom</groupId>
         <artifactId>metalus</artifactId>
-        <version>1.8.6-SNAPSHOT</version>
+        <version>1.8.7-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/pom.xml b/pom.xml
@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.acxiom</groupId>
     <artifactId>metalus</artifactId>
-    <version>1.8.6-SNAPSHOT</version>
+    <version>1.8.7-SNAPSHOT</version>
     <name>${project.artifactId}</name>
     <packaging>pom</packaging>
     <description>Metalus Pipeline Library</description>