diff --git a/go/tasks/v1/flytek8s/container_helper.go b/go/tasks/v1/flytek8s/container_helper.go
index 8f0d6597ac..93f92f97ab 100755
--- a/go/tasks/v1/flytek8s/container_helper.go
+++ b/go/tasks/v1/flytek8s/container_helper.go
@@ -29,19 +29,27 @@ func ApplyResourceOverrides(ctx context.Context, resources v1.ResourceRequiremen
 	if len(resources.Requests) == 0 {
 		resources.Requests = make(v1.ResourceList)
 	}
+	if len(resources.Limits) == 0 {
+		resources.Limits = make(v1.ResourceList)
+	}
+	
 	if _, found := resources.Requests[v1.ResourceCPU]; !found {
-		resources.Requests[v1.ResourceCPU] = resource.MustParse(config.GetK8sPluginConfig().DefaultCpuRequest)
+		// use cpu limit if set else default to config
+		if _, limitSet := resources.Limits[v1.ResourceCPU]; limitSet {
+			resources.Requests[v1.ResourceCPU] = resources.Limits[v1.ResourceCPU]
+		} else {
+			resources.Requests[v1.ResourceCPU] = resource.MustParse(config.GetK8sPluginConfig().DefaultCpuRequest)
+		}
 	}
 	if _, found := resources.Requests[v1.ResourceMemory]; !found {
-		resources.Requests[v1.ResourceMemory] = resource.MustParse(config.GetK8sPluginConfig().DefaultMemoryRequest)
+		// use memory limit if set else default to config
+		if _, limitSet := resources.Limits[v1.ResourceCPU]; limitSet {
+			resources.Requests[v1.ResourceMemory] = resources.Limits[v1.ResourceMemory]
+		} else {
+			resources.Requests[v1.ResourceMemory] = resource.MustParse(config.GetK8sPluginConfig().DefaultMemoryRequest)
+		}
 	}
 
-	if len(resources.Limits) == 0 {
-		resources.Limits = make(v1.ResourceList)
-	}
-	if len(resources.Requests) == 0 {
-		resources.Requests = make(v1.ResourceList)
-	}
 	if _, found := resources.Limits[v1.ResourceCPU]; !found {
 		logger.Infof(ctx, "found cpu limit missing, setting limit to the requested value %v", resources.Requests[v1.ResourceCPU])
 		resources.Limits[v1.ResourceCPU] = resources.Requests[v1.ResourceCPU]
diff --git a/go/tasks/v1/flytek8s/container_helper_test.go b/go/tasks/v1/flytek8s/container_helper_test.go
index c1f530152e..26a82b36bb 100755
--- a/go/tasks/v1/flytek8s/container_helper_test.go
+++ b/go/tasks/v1/flytek8s/container_helper_test.go
@@ -29,6 +29,14 @@ func TestApplyResourceOverrides_OverrideCpu(t *testing.T) {
 	})
 	assert.EqualValues(t, cpuRequest, overrides.Requests[v1.ResourceCPU])
 	assert.EqualValues(t, cpuLimit, overrides.Limits[v1.ResourceCPU])
+
+	// request equals limit if not set
+	overrides = ApplyResourceOverrides(context.Background(), v1.ResourceRequirements{
+		Limits: v1.ResourceList{
+			v1.ResourceCPU: cpuLimit,
+		},
+	})
+	assert.EqualValues(t, cpuLimit, overrides.Requests[v1.ResourceCPU])
 }
 
 func TestApplyResourceOverrides_OverrideMemory(t *testing.T) {
@@ -52,6 +60,17 @@ func TestApplyResourceOverrides_OverrideMemory(t *testing.T) {
 	})
 	assert.EqualValues(t, memoryRequest, overrides.Requests[v1.ResourceMemory])
 	assert.EqualValues(t, memoryLimit, overrides.Limits[v1.ResourceMemory])
+
+	// request equals limit if not set
+	cpuLimit := resource.MustParse("2")
+	overrides = ApplyResourceOverrides(context.Background(), v1.ResourceRequirements{
+		Limits: v1.ResourceList{
+			v1.ResourceMemory: memoryLimit,
+			v1.ResourceCPU:    cpuLimit,
+		},
+	})
+	assert.EqualValues(t, memoryLimit, overrides.Requests[v1.ResourceMemory])
+	assert.EqualValues(t, cpuLimit, overrides.Requests[v1.ResourceCPU])
 }
 
 func TestApplyResourceOverrides_RemoveStorage(t *testing.T) {