diff --git a/docs/deployment/configuration/generated/flyteadmin_config.rst b/docs/deployment/configuration/generated/flyteadmin_config.rst
index 492b5f4d89..e8464b1512 100644
--- a/docs/deployment/configuration/generated/flyteadmin_config.rst
+++ b/docs/deployment/configuration/generated/flyteadmin_config.rst
@@ -3522,13 +3522,13 @@ Enable events publishing to K8s events API.
 max-output-size-bytes (int64)
 ------------------------------------------------------------------------------------------------------------------------
 
-Maximum size of outputs per task
+Deprecated! Use storage.limits.maxDownloadMBs instead
 
 **Default Value**: 
 
 .. code-block:: yaml
 
-  "10485760"
+  "-1"
   
 
 enable-grpc-latency-metrics (bool)
diff --git a/docs/deployment/configuration/generated/flytepropeller_config.rst b/docs/deployment/configuration/generated/flytepropeller_config.rst
index db4f9a543e..f9473afffd 100644
--- a/docs/deployment/configuration/generated/flytepropeller_config.rst
+++ b/docs/deployment/configuration/generated/flytepropeller_config.rst
@@ -4166,13 +4166,13 @@ Enable events publishing to K8s events API.
 max-output-size-bytes (int64)
 ------------------------------------------------------------------------------------------------------------------------
 
-Maximum size of outputs per task
+Deprecated! Use storage.limits.maxDownloadMBs instead
 
 **Default Value**: 
 
 .. code-block:: yaml
 
-  "10485760"
+  "-1"
   
 
 enable-grpc-latency-metrics (bool)
diff --git a/docs/deployment/configuration/generated/scheduler_config.rst b/docs/deployment/configuration/generated/scheduler_config.rst
index 467d5f5d4e..47e4ee870a 100644
--- a/docs/deployment/configuration/generated/scheduler_config.rst
+++ b/docs/deployment/configuration/generated/scheduler_config.rst
@@ -3522,13 +3522,13 @@ Enable events publishing to K8s events API.
 max-output-size-bytes (int64)
 ------------------------------------------------------------------------------------------------------------------------
 
-Maximum size of outputs per task
+Deprecated! Use storage.limits.maxDownloadMBs instead
 
 **Default Value**: 
 
 .. code-block:: yaml
 
-  "10485760"
+  "-1"
   
 
 enable-grpc-latency-metrics (bool)
diff --git a/flytepropeller/pkg/controller/config/config.go b/flytepropeller/pkg/controller/config/config.go
index 698883fb48..89122da3ed 100644
--- a/flytepropeller/pkg/controller/config/config.go
+++ b/flytepropeller/pkg/controller/config/config.go
@@ -62,7 +62,7 @@ var (
 		GCInterval: config.Duration{
 			Duration: 30 * time.Minute,
 		},
-		MaxDatasetSizeBytes: 10 * 1024 * 1024,
+		MaxDatasetSizeBytes: -1,
 		Queue: CompositeQueueConfig{
 			Type: CompositeQueueBatch,
 			BatchingInterval: config.Duration{
@@ -142,7 +142,7 @@ type Config struct {
 	GCInterval               config.Duration      `json:"gc-interval" pflag:"Run periodic GC every 30 minutes"`
 	LeaderElection           LeaderElectionConfig `json:"leader-election,omitempty" pflag:",Config for leader election."`
 	PublishK8sEvents         bool                 `json:"publish-k8s-events" pflag:",Enable events publishing to K8s events API."`
-	MaxDatasetSizeBytes      int64                `json:"max-output-size-bytes" pflag:",Maximum size of outputs per task"`
+	MaxDatasetSizeBytes      int64                `json:"max-output-size-bytes" pflag:",Deprecated! Use storage.limits.maxDownloadMBs instead"`
 	EnableGrpcLatencyMetrics bool                 `json:"enable-grpc-latency-metrics" pflag:",Enable grpc latency metrics. Note Histograms metrics can be expensive on Prometheus servers."`
 	KubeConfig               KubeClientConfig     `json:"kube-client-config" pflag:",Configuration to control the Kubernetes client"`
 	NodeConfig               NodeConfig           `json:"node-config,omitempty" pflag:",config for a workflow node"`
diff --git a/flytepropeller/pkg/controller/config/config_flags.go b/flytepropeller/pkg/controller/config/config_flags.go
index c60f724ee2..b055aad558 100755
--- a/flytepropeller/pkg/controller/config/config_flags.go
+++ b/flytepropeller/pkg/controller/config/config_flags.go
@@ -85,7 +85,7 @@ func (cfg Config) GetPFlagSet(prefix string) *pflag.FlagSet {
 	cmdFlags.String(fmt.Sprintf("%v%v", prefix, "leader-election.renew-deadline"), defaultConfig.LeaderElection.RenewDeadline.String(), "Duration that the acting master will retry refreshing leadership before giving up.")
 	cmdFlags.String(fmt.Sprintf("%v%v", prefix, "leader-election.retry-period"), defaultConfig.LeaderElection.RetryPeriod.String(), "Duration the LeaderElector clients should wait between tries of actions.")
 	cmdFlags.Bool(fmt.Sprintf("%v%v", prefix, "publish-k8s-events"), defaultConfig.PublishK8sEvents, "Enable events publishing to K8s events API.")
-	cmdFlags.Int64(fmt.Sprintf("%v%v", prefix, "max-output-size-bytes"), defaultConfig.MaxDatasetSizeBytes, "Maximum size of outputs per task")
+	cmdFlags.Int64(fmt.Sprintf("%v%v", prefix, "max-output-size-bytes"), defaultConfig.MaxDatasetSizeBytes, "Deprecated! Use storage.limits.maxDownloadMBs instead")
 	cmdFlags.Bool(fmt.Sprintf("%v%v", prefix, "enable-grpc-latency-metrics"), defaultConfig.EnableGrpcLatencyMetrics, "Enable grpc latency metrics. Note Histograms metrics can be expensive on Prometheus servers.")
 	cmdFlags.Int(fmt.Sprintf("%v%v", prefix, "kube-client-config.burst"), defaultConfig.KubeConfig.Burst, "Max burst rate for throttle. 0 defaults to 10")
 	cmdFlags.String(fmt.Sprintf("%v%v", prefix, "kube-client-config.timeout"), defaultConfig.KubeConfig.Timeout.String(), "Max duration allowed for every request to KubeAPI before giving up. 0 implies no timeout.")
diff --git a/flytepropeller/pkg/controller/controller.go b/flytepropeller/pkg/controller/controller.go
index de28612c54..acd7747d3f 100644
--- a/flytepropeller/pkg/controller/controller.go
+++ b/flytepropeller/pkg/controller/controller.go
@@ -439,7 +439,7 @@ func New(ctx context.Context, cfg *config.Config, kubeClientset kubernetes.Inter
 	}
 
 	nodeExecutor, err := nodes.NewExecutor(ctx, cfg.NodeConfig, store, controller.enqueueWorkflowForNodeUpdates, eventSink,
-		launchPlanActor, launchPlanActor, cfg.MaxDatasetSizeBytes, storage.DataReference(cfg.DefaultRawOutputPrefix), kubeClient,
+		launchPlanActor, launchPlanActor, sCfg.Limits.GetLimitMegabytes*1024*1024, storage.DataReference(cfg.DefaultRawOutputPrefix), kubeClient,
 		catalogClient, recoveryClient, &cfg.EventConfig, cfg.ClusterID, signalClient, nodeHandlerFactory, scope)
 	if err != nil {
 		return nil, errors.Wrapf(err, "Failed to create Controller.")