fixing max parallelism (#594)

* fixing max parallelism Signed-off-by: Daniel Rammer <daniel@union.ai> * Run subset of functional tests Signed-off-by: eduardo apolinario <eapolinario@users.noreply.github.com> * Comment basics.deck.wf test out Signed-off-by: eduardo apolinario <eapolinario@users.noreply.github.com> * Use flytetools@master in e2e tests and bring decks test back Signed-off-by: eduardo apolinario <eapolinario@users.noreply.github.com> --------- Signed-off-by: Daniel Rammer <daniel@union.ai> Signed-off-by: eduardo apolinario <eapolinario@users.noreply.github.com> Co-authored-by: eduardo apolinario <eapolinario@users.noreply.github.com>
flyteorg · Jul 26, 2023 · 0fd75ed · 0fd75ed
1 parent 4322012
commit 0fd75ed
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/pkg/controller/nodes/task/handler.go b/pkg/controller/nodes/task/handler.go
@@ -668,6 +668,9 @@ func (t Handler) Handle(ctx context.Context, nCtx handler.NodeExecutionContext)
 			return handler.UnknownTransition, errors.Wrapf(errors.RuntimeExecutionError, nCtx.NodeID(), err, "failed during plugin execution")
 		}
 		if pluginTrns.IsPreviouslyObserved() {
+			if !pluginTrns.pInfo.Phase().IsTerminal() {
+				logger.Infof(ctx, "Parallelism now set to [%d].", nCtx.ExecutionContext().IncrementParallelism())
+			}
 			logger.Debugf(ctx, "No state change for Task, previously observed same transition. Short circuiting.")
 			return pluginTrns.FinalTransition(ctx)
 		}
@@ -755,8 +758,7 @@ func (t Handler) Handle(ctx context.Context, nCtx handler.NodeExecutionContext)
 	}
 
 	if !pluginTrns.pInfo.Phase().IsTerminal() {
-		eCtx := nCtx.ExecutionContext()
-		logger.Infof(ctx, "Parallelism now set to [%d].", eCtx.IncrementParallelism())
+		logger.Infof(ctx, "Parallelism now set to [%d].", nCtx.ExecutionContext().IncrementParallelism())
 	}
 	return pluginTrns.FinalTransition(ctx)
 }