diff --git a/ray-operator/controllers/ray/common/pod.go b/ray-operator/controllers/ray/common/pod.go
index 3d4d4d1d0cf..ad6037165fd 100644
--- a/ray-operator/controllers/ray/common/pod.go
+++ b/ray-operator/controllers/ray/common/pod.go
@@ -126,6 +126,7 @@ func configureGCSFaultTolerance(podTemplate *corev1.PodTemplateSpec, instance ra
 					Value: options.RedisAddress,
 				})
 				if options.RedisUsername != nil {
+					// Note that `redis-username` will be supported starting from Ray 2.41.
 					// If `GcsFaultToleranceOptions.RedisUsername` is set, it will be put into the
 					// `REDIS_USERNAME` environment variable later. Here, we use `$REDIS_USERNAME` in
 					// rayStartParams to refer to the environment variable.
@@ -148,17 +149,8 @@ func configureGCSFaultTolerance(podTemplate *corev1.PodTemplateSpec, instance ra
 					})
 				}
 			} else {
-				// If users directly set the `redis-username` in `rayStartParams` instead of referring
-				// to a K8s secret, we need to set the `REDIS_USERNAME` env var so that the Redis cleanup
-				// job can connect to Redis using the username.
-				if !utils.EnvVarExists(utils.REDIS_USERNAME, container.Env) {
-					// setting the REDIS_USERNAME env var from the params
-					redisUsernameEnv := corev1.EnvVar{Name: utils.REDIS_USERNAME}
-					if value, ok := instance.Spec.HeadGroupSpec.RayStartParams["redis-username"]; ok {
-						redisUsernameEnv.Value = value
-					}
-					container.Env = append(container.Env, redisUsernameEnv)
-				}
+				container.Env = append(container.Env, corev1.EnvVar{Name: utils.REDIS_USERNAME})
+				
 				// If users directly set the `redis-password` in `rayStartParams` instead of referring
 				// to a K8s secret, we need to set the `REDIS_PASSWORD` env var so that the Redis cleanup
 				// job can connect to Redis using the password. This is not recommended.
diff --git a/ray-operator/controllers/ray/common/pod_test.go b/ray-operator/controllers/ray/common/pod_test.go
index d8a22148640..38ff5003099 100644
--- a/ray-operator/controllers/ray/common/pod_test.go
+++ b/ray-operator/controllers/ray/common/pod_test.go
@@ -518,6 +518,19 @@ func TestConfigureGCSFaultToleranceWithGcsFTOptions(t *testing.T) {
 			},
 			isHeadPod: true,
 		},
+		{
+			name: "GCS FT enabled with redis username and password",
+			gcsFTOptions: &rayv1.GcsFaultToleranceOptions{
+				RedisAddress: "redis:6379",
+				RedisUsername: &rayv1.RedisCredential{
+					Value: "test-username",
+				},
+				RedisPassword: &rayv1.RedisCredential{
+					Value: "test-password",
+				},
+			},
+			isHeadPod: true,
+		},
 		{
 			name: "GCS FT enabled with redis password in secret",
 			gcsFTOptions: &rayv1.GcsFaultToleranceOptions{
@@ -533,7 +546,28 @@ func TestConfigureGCSFaultToleranceWithGcsFTOptions(t *testing.T) {
 			isHeadPod: true,
 		},
 		{
-			name: "GCS FT enabled with redis password in secret",
+			name: "GCS FT enabled with redis username and password in secret",
+			gcsFTOptions: &rayv1.GcsFaultToleranceOptions{
+				RedisAddress: "redis:6379",
+				RedisUsername: &rayv1.RedisCredential{
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{
+							FieldPath: "spec.redisUsername",
+						},
+					},
+				},
+				RedisPassword: &rayv1.RedisCredential{
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{
+							FieldPath: "spec.redisPassword",
+						},
+					},
+				},
+			},
+			isHeadPod: true,
+		},
+		{
+			name: "GCS FT enabled with redis external storage namespace",
 			gcsFTOptions: &rayv1.GcsFaultToleranceOptions{
 				RedisAddress:             "redis:6379",
 				ExternalStorageNamespace: "test-ns",
@@ -593,6 +627,12 @@ func TestConfigureGCSFaultToleranceWithGcsFTOptions(t *testing.T) {
 				env := getEnvVar(container, utils.RAY_REDIS_ADDRESS)
 				assert.Equal(t, env.Value, "redis:6379")
 
+				if test.gcsFTOptions.RedisUsername != nil {
+					env := getEnvVar(container, utils.REDIS_USERNAME)
+					assert.Equal(t, env.Value, test.gcsFTOptions.RedisUsername.Value)
+					assert.Equal(t, env.ValueFrom, test.gcsFTOptions.RedisUsername.ValueFrom)
+				}
+
 				if test.gcsFTOptions.RedisPassword != nil {
 					env := getEnvVar(container, utils.REDIS_PASSWORD)
 					assert.Equal(t, env.Value, test.gcsFTOptions.RedisPassword.Value)
diff --git a/ray-operator/controllers/ray/raycluster_controller.go b/ray-operator/controllers/ray/raycluster_controller.go
index d69e1ee4ad2..8cf80089b97 100644
--- a/ray-operator/controllers/ray/raycluster_controller.go
+++ b/ray-operator/controllers/ray/raycluster_controller.go
@@ -1237,7 +1237,7 @@ func (r *RayClusterReconciler) buildRedisCleanupJob(ctx context.Context, instanc
 			"parsed = urlparse(redis_address); ",
 	}
 	if utils.EnvVarExists(utils.REDIS_USERNAME, pod.Spec.Containers[utils.RayContainerIndex].Env) {
-		pod.Spec.Containers[utils.RayContainerIndex].Args[0] += "sys.exit(1) if not cleanup_redis_storage(host=parsed.hostname, port=parsed.port, username=os.getenv('REDIS_USERNAME', parsed.username), password=os.getenv('REDIS_PASSWORD', parsed.password), use_ssl=parsed.scheme=='rediss', storage_namespace=os.getenv('RAY_external_storage_namespace')) else None\""
+		pod.Spec.Containers[utils.RayContainerIndex].Args[0] += "sys.exit(1) if not cleanup_redis_storage(host=parsed.hostname, port=parsed.port, username=os.getenv('REDIS_USERNAME', parsed.username), password=os.getenv('REDIS_PASSWORD', parsed.password or ''), use_ssl=parsed.scheme=='rediss', storage_namespace=os.getenv('RAY_external_storage_namespace')) else None\""
 	} else {
 		pod.Spec.Containers[utils.RayContainerIndex].Args[0] += "sys.exit(1) if not cleanup_redis_storage(host=parsed.hostname, port=parsed.port, password=os.getenv('REDIS_PASSWORD', parsed.password or ''), use_ssl=parsed.scheme=='rediss', storage_namespace=os.getenv('RAY_external_storage_namespace')) else None\""
 	}