fixing max parallelism (flyteorg#594)

* fixing max parallelism Signed-off-by: Daniel Rammer <[email protected]> * Run subset of functional tests Signed-off-by: eduardo apolinario <[email protected]> * Comment basics.deck.wf test out Signed-off-by: eduardo apolinario <[email protected]> * Use flytetools@master in e2e tests and bring decks test back Signed-off-by: eduardo apolinario <[email protected]> --------- Signed-off-by: Daniel Rammer <[email protected]> Signed-off-by: eduardo apolinario <[email protected]> Co-authored-by: eduardo apolinario <[email protected]>
eapolinario · Jul 26, 2023 · 405bd6f · 405bd6f
1 parent d60d711
commit 405bd6f
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/flytepropeller/pkg/controller/nodes/task/handler.go b/flytepropeller/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)
 }