diff --git a/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.cc b/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.cc
index fc5ea74702..7579e59cc0 100644
--- a/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.cc
+++ b/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.cc
@@ -157,6 +157,8 @@ const ::google::protobuf::uint32 TableStruct_flyteidl_2fplugins_2fray_2eproto::o
   ~0u,  // no _weak_field_map_
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayJob, ray_cluster_),
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayJob, runtime_env_),
+  PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayJob, shutdown_after_job_finishes_),
+  PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayJob, ttl_seconds_after_finished_),
   ~0u,  // no _has_bits_
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayCluster, _internal_metadata_),
   ~0u,  // no _extensions_
@@ -164,6 +166,7 @@ const ::google::protobuf::uint32 TableStruct_flyteidl_2fplugins_2fray_2eproto::o
   ~0u,  // no _weak_field_map_
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayCluster, head_group_spec_),
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayCluster, worker_group_spec_),
+  PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::RayCluster, enable_autoscaling_),
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::HeadGroupSpec_RayStartParamsEntry_DoNotUse, _has_bits_),
   PROTOBUF_FIELD_OFFSET(::flyteidl::plugins::HeadGroupSpec_RayStartParamsEntry_DoNotUse, _internal_metadata_),
   ~0u,  // no _extensions_
@@ -201,11 +204,11 @@ const ::google::protobuf::uint32 TableStruct_flyteidl_2fplugins_2fray_2eproto::o
 };
 static const ::google::protobuf::internal::MigrationSchema schemas[] PROTOBUF_SECTION_VARIABLE(protodesc_cold) = {
   { 0, -1, sizeof(::flyteidl::plugins::RayJob)},
-  { 7, -1, sizeof(::flyteidl::plugins::RayCluster)},
-  { 14, 21, sizeof(::flyteidl::plugins::HeadGroupSpec_RayStartParamsEntry_DoNotUse)},
-  { 23, -1, sizeof(::flyteidl::plugins::HeadGroupSpec)},
-  { 29, 36, sizeof(::flyteidl::plugins::WorkerGroupSpec_RayStartParamsEntry_DoNotUse)},
-  { 38, -1, sizeof(::flyteidl::plugins::WorkerGroupSpec)},
+  { 9, -1, sizeof(::flyteidl::plugins::RayCluster)},
+  { 17, 24, sizeof(::flyteidl::plugins::HeadGroupSpec_RayStartParamsEntry_DoNotUse)},
+  { 26, -1, sizeof(::flyteidl::plugins::HeadGroupSpec)},
+  { 32, 39, sizeof(::flyteidl::plugins::WorkerGroupSpec_RayStartParamsEntry_DoNotUse)},
+  { 41, -1, sizeof(::flyteidl::plugins::WorkerGroupSpec)},
 };
 
 static ::google::protobuf::Message const * const file_default_instances[] = {
@@ -225,29 +228,31 @@ ::google::protobuf::internal::AssignDescriptorsTable assign_descriptors_table_fl
 
 const char descriptor_table_protodef_flyteidl_2fplugins_2fray_2eproto[] =
   "\n\032flyteidl/plugins/ray.proto\022\020flyteidl.p"
-  "lugins\"P\n\006RayJob\0221\n\013ray_cluster\030\001 \001(\0132\034."
-  "flyteidl.plugins.RayCluster\022\023\n\013runtime_e"
-  "nv\030\002 \001(\t\"\204\001\n\nRayCluster\0228\n\017head_group_sp"
-  "ec\030\001 \001(\0132\037.flyteidl.plugins.HeadGroupSpe"
-  "c\022<\n\021worker_group_spec\030\002 \003(\0132!.flyteidl."
-  "plugins.WorkerGroupSpec\"\225\001\n\rHeadGroupSpe"
-  "c\022M\n\020ray_start_params\030\001 \003(\01323.flyteidl.p"
-  "lugins.HeadGroupSpec.RayStartParamsEntry"
-  "\0325\n\023RayStartParamsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005"
-  "value\030\002 \001(\t:\0028\001\"\353\001\n\017WorkerGroupSpec\022\022\n\ng"
-  "roup_name\030\001 \001(\t\022\020\n\010replicas\030\002 \001(\005\022\024\n\014min"
-  "_replicas\030\003 \001(\005\022\024\n\014max_replicas\030\004 \001(\005\022O\n"
-  "\020ray_start_params\030\005 \003(\01325.flyteidl.plugi"
-  "ns.WorkerGroupSpec.RayStartParamsEntry\0325"
-  "\n\023RayStartParamsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005va"
-  "lue\030\002 \001(\t:\0028\001B\?Z=github.com/flyteorg/fly"
-  "te/flyteidl/gen/pb-go/flyteidl/pluginsb\006"
-  "proto3"
+  "lugins\"\231\001\n\006RayJob\0221\n\013ray_cluster\030\001 \001(\0132\034"
+  ".flyteidl.plugins.RayCluster\022\023\n\013runtime_"
+  "env\030\002 \001(\t\022#\n\033shutdown_after_job_finishes"
+  "\030\003 \001(\010\022\"\n\032ttl_seconds_after_finished\030\004 \001"
+  "(\005\"\240\001\n\nRayCluster\0228\n\017head_group_spec\030\001 \001"
+  "(\0132\037.flyteidl.plugins.HeadGroupSpec\022<\n\021w"
+  "orker_group_spec\030\002 \003(\0132!.flyteidl.plugin"
+  "s.WorkerGroupSpec\022\032\n\022enable_autoscaling\030"
+  "\003 \001(\010\"\225\001\n\rHeadGroupSpec\022M\n\020ray_start_par"
+  "ams\030\001 \003(\01323.flyteidl.plugins.HeadGroupSp"
+  "ec.RayStartParamsEntry\0325\n\023RayStartParams"
+  "Entry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\353\001"
+  "\n\017WorkerGroupSpec\022\022\n\ngroup_name\030\001 \001(\t\022\020\n"
+  "\010replicas\030\002 \001(\005\022\024\n\014min_replicas\030\003 \001(\005\022\024\n"
+  "\014max_replicas\030\004 \001(\005\022O\n\020ray_start_params\030"
+  "\005 \003(\01325.flyteidl.plugins.WorkerGroupSpec"
+  ".RayStartParamsEntry\0325\n\023RayStartParamsEn"
+  "try\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001B\?Z=g"
+  "ithub.com/flyteorg/flyte/flyteidl/gen/pb"
+  "-go/flyteidl/pluginsb\006proto3"
   ;
 ::google::protobuf::internal::DescriptorTable descriptor_table_flyteidl_2fplugins_2fray_2eproto = {
   false, InitDefaults_flyteidl_2fplugins_2fray_2eproto, 
   descriptor_table_protodef_flyteidl_2fplugins_2fray_2eproto,
-  "flyteidl/plugins/ray.proto", &assign_descriptors_table_flyteidl_2fplugins_2fray_2eproto, 726,
+  "flyteidl/plugins/ray.proto", &assign_descriptors_table_flyteidl_2fplugins_2fray_2eproto, 828,
 };
 
 void AddDescriptors_flyteidl_2fplugins_2fray_2eproto() {
@@ -280,6 +285,8 @@ RayJob::HasBitSetters::ray_cluster(const RayJob* msg) {
 #if !defined(_MSC_VER) || _MSC_VER >= 1900
 const int RayJob::kRayClusterFieldNumber;
 const int RayJob::kRuntimeEnvFieldNumber;
+const int RayJob::kShutdownAfterJobFinishesFieldNumber;
+const int RayJob::kTtlSecondsAfterFinishedFieldNumber;
 #endif  // !defined(_MSC_VER) || _MSC_VER >= 1900
 
 RayJob::RayJob()
@@ -300,6 +307,9 @@ RayJob::RayJob(const RayJob& from)
   } else {
     ray_cluster_ = nullptr;
   }
+  ::memcpy(&shutdown_after_job_finishes_, &from.shutdown_after_job_finishes_,
+    static_cast<size_t>(reinterpret_cast<char*>(&ttl_seconds_after_finished_) -
+    reinterpret_cast<char*>(&shutdown_after_job_finishes_)) + sizeof(ttl_seconds_after_finished_));
   // @@protoc_insertion_point(copy_constructor:flyteidl.plugins.RayJob)
 }
 
@@ -307,7 +317,9 @@ void RayJob::SharedCtor() {
   ::google::protobuf::internal::InitSCC(
       &scc_info_RayJob_flyteidl_2fplugins_2fray_2eproto.base);
   runtime_env_.UnsafeSetDefault(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
-  ray_cluster_ = nullptr;
+  ::memset(&ray_cluster_, 0, static_cast<size_t>(
+      reinterpret_cast<char*>(&ttl_seconds_after_finished_) -
+      reinterpret_cast<char*>(&ray_cluster_)) + sizeof(ttl_seconds_after_finished_));
 }
 
 RayJob::~RayJob() {
@@ -340,6 +352,9 @@ void RayJob::Clear() {
     delete ray_cluster_;
   }
   ray_cluster_ = nullptr;
+  ::memset(&shutdown_after_job_finishes_, 0, static_cast<size_t>(
+      reinterpret_cast<char*>(&ttl_seconds_after_finished_) -
+      reinterpret_cast<char*>(&shutdown_after_job_finishes_)) + sizeof(ttl_seconds_after_finished_));
   _internal_metadata_.Clear();
 }
 
@@ -385,6 +400,20 @@ const char* RayJob::_InternalParse(const char* begin, const char* end, void* obj
         ptr += size;
         break;
       }
+      // bool shutdown_after_job_finishes = 3;
+      case 3: {
+        if (static_cast<::google::protobuf::uint8>(tag) != 24) goto handle_unusual;
+        msg->set_shutdown_after_job_finishes(::google::protobuf::internal::ReadVarint(&ptr));
+        GOOGLE_PROTOBUF_PARSER_ASSERT(ptr);
+        break;
+      }
+      // int32 ttl_seconds_after_finished = 4;
+      case 4: {
+        if (static_cast<::google::protobuf::uint8>(tag) != 32) goto handle_unusual;
+        msg->set_ttl_seconds_after_finished(::google::protobuf::internal::ReadVarint(&ptr));
+        GOOGLE_PROTOBUF_PARSER_ASSERT(ptr);
+        break;
+      }
       default: {
       handle_unusual:
         if ((tag & 7) == 4 || tag == 0) {
@@ -445,6 +474,32 @@ bool RayJob::MergePartialFromCodedStream(
         break;
       }
 
+      // bool shutdown_after_job_finishes = 3;
+      case 3: {
+        if (static_cast< ::google::protobuf::uint8>(tag) == (24 & 0xFF)) {
+
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   bool, ::google::protobuf::internal::WireFormatLite::TYPE_BOOL>(
+                 input, &shutdown_after_job_finishes_)));
+        } else {
+          goto handle_unusual;
+        }
+        break;
+      }
+
+      // int32 ttl_seconds_after_finished = 4;
+      case 4: {
+        if (static_cast< ::google::protobuf::uint8>(tag) == (32 & 0xFF)) {
+
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 input, &ttl_seconds_after_finished_)));
+        } else {
+          goto handle_unusual;
+        }
+        break;
+      }
+
       default: {
       handle_unusual:
         if (tag == 0) {
@@ -488,6 +543,16 @@ void RayJob::SerializeWithCachedSizes(
       2, this->runtime_env(), output);
   }
 
+  // bool shutdown_after_job_finishes = 3;
+  if (this->shutdown_after_job_finishes() != 0) {
+    ::google::protobuf::internal::WireFormatLite::WriteBool(3, this->shutdown_after_job_finishes(), output);
+  }
+
+  // int32 ttl_seconds_after_finished = 4;
+  if (this->ttl_seconds_after_finished() != 0) {
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(4, this->ttl_seconds_after_finished(), output);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
         _internal_metadata_.unknown_fields(), output);
@@ -519,6 +584,16 @@ ::google::protobuf::uint8* RayJob::InternalSerializeWithCachedSizesToArray(
         2, this->runtime_env(), target);
   }
 
+  // bool shutdown_after_job_finishes = 3;
+  if (this->shutdown_after_job_finishes() != 0) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(3, this->shutdown_after_job_finishes(), target);
+  }
+
+  // int32 ttl_seconds_after_finished = 4;
+  if (this->ttl_seconds_after_finished() != 0) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(4, this->ttl_seconds_after_finished(), target);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
         _internal_metadata_.unknown_fields(), target);
@@ -554,6 +629,18 @@ size_t RayJob::ByteSizeLong() const {
         *ray_cluster_);
   }
 
+  // bool shutdown_after_job_finishes = 3;
+  if (this->shutdown_after_job_finishes() != 0) {
+    total_size += 1 + 1;
+  }
+
+  // int32 ttl_seconds_after_finished = 4;
+  if (this->ttl_seconds_after_finished() != 0) {
+    total_size += 1 +
+      ::google::protobuf::internal::WireFormatLite::Int32Size(
+        this->ttl_seconds_after_finished());
+  }
+
   int cached_size = ::google::protobuf::internal::ToCachedSize(total_size);
   SetCachedSize(cached_size);
   return total_size;
@@ -588,6 +675,12 @@ void RayJob::MergeFrom(const RayJob& from) {
   if (from.has_ray_cluster()) {
     mutable_ray_cluster()->::flyteidl::plugins::RayCluster::MergeFrom(from.ray_cluster());
   }
+  if (from.shutdown_after_job_finishes() != 0) {
+    set_shutdown_after_job_finishes(from.shutdown_after_job_finishes());
+  }
+  if (from.ttl_seconds_after_finished() != 0) {
+    set_ttl_seconds_after_finished(from.ttl_seconds_after_finished());
+  }
 }
 
 void RayJob::CopyFrom(const ::google::protobuf::Message& from) {
@@ -618,6 +711,8 @@ void RayJob::InternalSwap(RayJob* other) {
   runtime_env_.Swap(&other->runtime_env_, &::google::protobuf::internal::GetEmptyStringAlreadyInited(),
     GetArenaNoVirtual());
   swap(ray_cluster_, other->ray_cluster_);
+  swap(shutdown_after_job_finishes_, other->shutdown_after_job_finishes_);
+  swap(ttl_seconds_after_finished_, other->ttl_seconds_after_finished_);
 }
 
 ::google::protobuf::Metadata RayJob::GetMetadata() const {
@@ -644,6 +739,7 @@ RayCluster::HasBitSetters::head_group_spec(const RayCluster* msg) {
 #if !defined(_MSC_VER) || _MSC_VER >= 1900
 const int RayCluster::kHeadGroupSpecFieldNumber;
 const int RayCluster::kWorkerGroupSpecFieldNumber;
+const int RayCluster::kEnableAutoscalingFieldNumber;
 #endif  // !defined(_MSC_VER) || _MSC_VER >= 1900
 
 RayCluster::RayCluster()
@@ -661,13 +757,16 @@ RayCluster::RayCluster(const RayCluster& from)
   } else {
     head_group_spec_ = nullptr;
   }
+  enable_autoscaling_ = from.enable_autoscaling_;
   // @@protoc_insertion_point(copy_constructor:flyteidl.plugins.RayCluster)
 }
 
 void RayCluster::SharedCtor() {
   ::google::protobuf::internal::InitSCC(
       &scc_info_RayCluster_flyteidl_2fplugins_2fray_2eproto.base);
-  head_group_spec_ = nullptr;
+  ::memset(&head_group_spec_, 0, static_cast<size_t>(
+      reinterpret_cast<char*>(&enable_autoscaling_) -
+      reinterpret_cast<char*>(&head_group_spec_)) + sizeof(enable_autoscaling_));
 }
 
 RayCluster::~RayCluster() {
@@ -699,6 +798,7 @@ void RayCluster::Clear() {
     delete head_group_spec_;
   }
   head_group_spec_ = nullptr;
+  enable_autoscaling_ = false;
   _internal_metadata_.Clear();
 }
 
@@ -744,6 +844,13 @@ const char* RayCluster::_InternalParse(const char* begin, const char* end, void*
         } while ((::google::protobuf::io::UnalignedLoad<::google::protobuf::uint64>(ptr) & 255) == 18 && (ptr += 1));
         break;
       }
+      // bool enable_autoscaling = 3;
+      case 3: {
+        if (static_cast<::google::protobuf::uint8>(tag) != 24) goto handle_unusual;
+        msg->set_enable_autoscaling(::google::protobuf::internal::ReadVarint(&ptr));
+        GOOGLE_PROTOBUF_PARSER_ASSERT(ptr);
+        break;
+      }
       default: {
       handle_unusual:
         if ((tag & 7) == 4 || tag == 0) {
@@ -796,6 +903,19 @@ bool RayCluster::MergePartialFromCodedStream(
         break;
       }
 
+      // bool enable_autoscaling = 3;
+      case 3: {
+        if (static_cast< ::google::protobuf::uint8>(tag) == (24 & 0xFF)) {
+
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   bool, ::google::protobuf::internal::WireFormatLite::TYPE_BOOL>(
+                 input, &enable_autoscaling_)));
+        } else {
+          goto handle_unusual;
+        }
+        break;
+      }
+
       default: {
       handle_unusual:
         if (tag == 0) {
@@ -838,6 +958,11 @@ void RayCluster::SerializeWithCachedSizes(
       output);
   }
 
+  // bool enable_autoscaling = 3;
+  if (this->enable_autoscaling() != 0) {
+    ::google::protobuf::internal::WireFormatLite::WriteBool(3, this->enable_autoscaling(), output);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
         _internal_metadata_.unknown_fields(), output);
@@ -866,6 +991,11 @@ ::google::protobuf::uint8* RayCluster::InternalSerializeWithCachedSizesToArray(
         2, this->worker_group_spec(static_cast<int>(i)), target);
   }
 
+  // bool enable_autoscaling = 3;
+  if (this->enable_autoscaling() != 0) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(3, this->enable_autoscaling(), target);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
         _internal_metadata_.unknown_fields(), target);
@@ -905,6 +1035,11 @@ size_t RayCluster::ByteSizeLong() const {
         *head_group_spec_);
   }
 
+  // bool enable_autoscaling = 3;
+  if (this->enable_autoscaling() != 0) {
+    total_size += 1 + 1;
+  }
+
   int cached_size = ::google::protobuf::internal::ToCachedSize(total_size);
   SetCachedSize(cached_size);
   return total_size;
@@ -936,6 +1071,9 @@ void RayCluster::MergeFrom(const RayCluster& from) {
   if (from.has_head_group_spec()) {
     mutable_head_group_spec()->::flyteidl::plugins::HeadGroupSpec::MergeFrom(from.head_group_spec());
   }
+  if (from.enable_autoscaling() != 0) {
+    set_enable_autoscaling(from.enable_autoscaling());
+  }
 }
 
 void RayCluster::CopyFrom(const ::google::protobuf::Message& from) {
@@ -965,6 +1103,7 @@ void RayCluster::InternalSwap(RayCluster* other) {
   _internal_metadata_.Swap(&other->_internal_metadata_);
   CastToBase(&worker_group_spec_)->InternalSwap(CastToBase(&other->worker_group_spec_));
   swap(head_group_spec_, other->head_group_spec_);
+  swap(enable_autoscaling_, other->enable_autoscaling_);
 }
 
 ::google::protobuf::Metadata RayCluster::GetMetadata() const {
diff --git a/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.h b/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.h
index 5de7bb122c..6d49906b97 100644
--- a/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.h
+++ b/flyteidl/gen/pb-cpp/flyteidl/plugins/ray.pb.h
@@ -206,6 +206,18 @@ class RayJob final :
   ::flyteidl::plugins::RayCluster* mutable_ray_cluster();
   void set_allocated_ray_cluster(::flyteidl::plugins::RayCluster* ray_cluster);
 
+  // bool shutdown_after_job_finishes = 3;
+  void clear_shutdown_after_job_finishes();
+  static const int kShutdownAfterJobFinishesFieldNumber = 3;
+  bool shutdown_after_job_finishes() const;
+  void set_shutdown_after_job_finishes(bool value);
+
+  // int32 ttl_seconds_after_finished = 4;
+  void clear_ttl_seconds_after_finished();
+  static const int kTtlSecondsAfterFinishedFieldNumber = 4;
+  ::google::protobuf::int32 ttl_seconds_after_finished() const;
+  void set_ttl_seconds_after_finished(::google::protobuf::int32 value);
+
   // @@protoc_insertion_point(class_scope:flyteidl.plugins.RayJob)
  private:
   class HasBitSetters;
@@ -213,6 +225,8 @@ class RayJob final :
   ::google::protobuf::internal::InternalMetadataWithArena _internal_metadata_;
   ::google::protobuf::internal::ArenaStringPtr runtime_env_;
   ::flyteidl::plugins::RayCluster* ray_cluster_;
+  bool shutdown_after_job_finishes_;
+  ::google::protobuf::int32 ttl_seconds_after_finished_;
   mutable ::google::protobuf::internal::CachedSize _cached_size_;
   friend struct ::TableStruct_flyteidl_2fplugins_2fray_2eproto;
 };
@@ -334,6 +348,12 @@ class RayCluster final :
   ::flyteidl::plugins::HeadGroupSpec* mutable_head_group_spec();
   void set_allocated_head_group_spec(::flyteidl::plugins::HeadGroupSpec* head_group_spec);
 
+  // bool enable_autoscaling = 3;
+  void clear_enable_autoscaling();
+  static const int kEnableAutoscalingFieldNumber = 3;
+  bool enable_autoscaling() const;
+  void set_enable_autoscaling(bool value);
+
   // @@protoc_insertion_point(class_scope:flyteidl.plugins.RayCluster)
  private:
   class HasBitSetters;
@@ -341,6 +361,7 @@ class RayCluster final :
   ::google::protobuf::internal::InternalMetadataWithArena _internal_metadata_;
   ::google::protobuf::RepeatedPtrField< ::flyteidl::plugins::WorkerGroupSpec > worker_group_spec_;
   ::flyteidl::plugins::HeadGroupSpec* head_group_spec_;
+  bool enable_autoscaling_;
   mutable ::google::protobuf::internal::CachedSize _cached_size_;
   friend struct ::TableStruct_flyteidl_2fplugins_2fray_2eproto;
 };
@@ -785,6 +806,34 @@ inline void RayJob::set_allocated_runtime_env(::std::string* runtime_env) {
   // @@protoc_insertion_point(field_set_allocated:flyteidl.plugins.RayJob.runtime_env)
 }
 
+// bool shutdown_after_job_finishes = 3;
+inline void RayJob::clear_shutdown_after_job_finishes() {
+  shutdown_after_job_finishes_ = false;
+}
+inline bool RayJob::shutdown_after_job_finishes() const {
+  // @@protoc_insertion_point(field_get:flyteidl.plugins.RayJob.shutdown_after_job_finishes)
+  return shutdown_after_job_finishes_;
+}
+inline void RayJob::set_shutdown_after_job_finishes(bool value) {
+  
+  shutdown_after_job_finishes_ = value;
+  // @@protoc_insertion_point(field_set:flyteidl.plugins.RayJob.shutdown_after_job_finishes)
+}
+
+// int32 ttl_seconds_after_finished = 4;
+inline void RayJob::clear_ttl_seconds_after_finished() {
+  ttl_seconds_after_finished_ = 0;
+}
+inline ::google::protobuf::int32 RayJob::ttl_seconds_after_finished() const {
+  // @@protoc_insertion_point(field_get:flyteidl.plugins.RayJob.ttl_seconds_after_finished)
+  return ttl_seconds_after_finished_;
+}
+inline void RayJob::set_ttl_seconds_after_finished(::google::protobuf::int32 value) {
+  
+  ttl_seconds_after_finished_ = value;
+  // @@protoc_insertion_point(field_set:flyteidl.plugins.RayJob.ttl_seconds_after_finished)
+}
+
 // -------------------------------------------------------------------
 
 // RayCluster
@@ -870,6 +919,20 @@ RayCluster::worker_group_spec() const {
   return worker_group_spec_;
 }
 
+// bool enable_autoscaling = 3;
+inline void RayCluster::clear_enable_autoscaling() {
+  enable_autoscaling_ = false;
+}
+inline bool RayCluster::enable_autoscaling() const {
+  // @@protoc_insertion_point(field_get:flyteidl.plugins.RayCluster.enable_autoscaling)
+  return enable_autoscaling_;
+}
+inline void RayCluster::set_enable_autoscaling(bool value) {
+  
+  enable_autoscaling_ = value;
+  // @@protoc_insertion_point(field_set:flyteidl.plugins.RayCluster.enable_autoscaling)
+}
+
 // -------------------------------------------------------------------
 
 // -------------------------------------------------------------------
diff --git a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
index f19e755f85..b092b143cc 100644
--- a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
+++ b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
@@ -26,10 +26,14 @@ type RayJob struct {
 	RayCluster *RayCluster `protobuf:"bytes,1,opt,name=ray_cluster,json=rayCluster,proto3" json:"ray_cluster,omitempty"`
 	// runtime_env is base64 encoded.
 	// Ray runtime environments: https://docs.ray.io/en/latest/ray-core/handling-dependencies.html#runtime-environments
-	RuntimeEnv           string   `protobuf:"bytes,2,opt,name=runtime_env,json=runtimeEnv,proto3" json:"runtime_env,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
+	RuntimeEnv string `protobuf:"bytes,2,opt,name=runtime_env,json=runtimeEnv,proto3" json:"runtime_env,omitempty"`
+	// shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+	ShutdownAfterJobFinishes bool `protobuf:"varint,3,opt,name=shutdown_after_job_finishes,json=shutdownAfterJobFinishes,proto3" json:"shutdown_after_job_finishes,omitempty"`
+	// ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+	TtlSecondsAfterFinished int32    `protobuf:"varint,4,opt,name=ttl_seconds_after_finished,json=ttlSecondsAfterFinished,proto3" json:"ttl_seconds_after_finished,omitempty"`
+	XXX_NoUnkeyedLiteral    struct{} `json:"-"`
+	XXX_unrecognized        []byte   `json:"-"`
+	XXX_sizecache           int32    `json:"-"`
 }
 
 func (m *RayJob) Reset()         { *m = RayJob{} }
@@ -71,15 +75,31 @@ func (m *RayJob) GetRuntimeEnv() string {
 	return ""
 }
 
+func (m *RayJob) GetShutdownAfterJobFinishes() bool {
+	if m != nil {
+		return m.ShutdownAfterJobFinishes
+	}
+	return false
+}
+
+func (m *RayJob) GetTtlSecondsAfterFinished() int32 {
+	if m != nil {
+		return m.TtlSecondsAfterFinished
+	}
+	return 0
+}
+
 // Define Ray cluster defines the desired state of RayCluster
 type RayCluster struct {
 	// HeadGroupSpecs are the spec for the head pod
 	HeadGroupSpec *HeadGroupSpec `protobuf:"bytes,1,opt,name=head_group_spec,json=headGroupSpec,proto3" json:"head_group_spec,omitempty"`
 	// WorkerGroupSpecs are the specs for the worker pods
-	WorkerGroupSpec      []*WorkerGroupSpec `protobuf:"bytes,2,rep,name=worker_group_spec,json=workerGroupSpec,proto3" json:"worker_group_spec,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
-	XXX_unrecognized     []byte             `json:"-"`
-	XXX_sizecache        int32              `json:"-"`
+	WorkerGroupSpec []*WorkerGroupSpec `protobuf:"bytes,2,rep,name=worker_group_spec,json=workerGroupSpec,proto3" json:"worker_group_spec,omitempty"`
+	// Whether to enable autoscaling.
+	EnableAutoscaling    bool     `protobuf:"varint,3,opt,name=enable_autoscaling,json=enableAutoscaling,proto3" json:"enable_autoscaling,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
 }
 
 func (m *RayCluster) Reset()         { *m = RayCluster{} }
@@ -121,6 +141,13 @@ func (m *RayCluster) GetWorkerGroupSpec() []*WorkerGroupSpec {
 	return nil
 }
 
+func (m *RayCluster) GetEnableAutoscaling() bool {
+	if m != nil {
+		return m.EnableAutoscaling
+	}
+	return false
+}
+
 // HeadGroupSpec are the spec for the head pod
 type HeadGroupSpec struct {
 	// Optional. RayStartParams are the params of the start command: address, object-store-memory.
@@ -253,31 +280,37 @@ func init() {
 func init() { proto.RegisterFile("flyteidl/plugins/ray.proto", fileDescriptor_b980f6d58c7489d7) }
 
 var fileDescriptor_b980f6d58c7489d7 = []byte{
-	// 413 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x93, 0xcd, 0xca, 0xd3, 0x40,
-	0x14, 0x86, 0x49, 0x6a, 0x3f, 0xec, 0x89, 0x9f, 0xad, 0xa3, 0x8b, 0x52, 0x94, 0xb6, 0x59, 0x75,
-	0x63, 0x02, 0x2d, 0x82, 0x08, 0x45, 0x54, 0x4a, 0x45, 0x50, 0x64, 0xba, 0x10, 0x04, 0x09, 0x93,
-	0x74, 0x4c, 0x42, 0x93, 0x99, 0x61, 0x32, 0x69, 0x9b, 0xfb, 0xf1, 0x06, 0xdc, 0x78, 0x7d, 0x92,
-	0x49, 0x4c, 0x7f, 0xa1, 0x3b, 0x77, 0xe7, 0xe7, 0x3d, 0xef, 0x99, 0x3c, 0xe1, 0xc0, 0xe0, 0x67,
-	0x52, 0x28, 0x1a, 0xaf, 0x13, 0x57, 0x24, 0x79, 0x18, 0xb3, 0xcc, 0x95, 0xa4, 0x70, 0x84, 0xe4,
-	0x8a, 0xa3, 0xde, 0xbf, 0x9e, 0x53, 0xf7, 0xec, 0x08, 0xee, 0x30, 0x29, 0x3e, 0x71, 0x1f, 0xcd,
-	0xc1, 0x92, 0xa4, 0xf0, 0x82, 0x24, 0xcf, 0x14, 0x95, 0x7d, 0x63, 0x64, 0x4c, 0xac, 0xe9, 0x73,
-	0xe7, 0x7c, 0xc2, 0xc1, 0xa4, 0xf8, 0x50, 0x69, 0x30, 0xc8, 0x26, 0x46, 0x43, 0xb0, 0x64, 0xce,
-	0x54, 0x9c, 0x52, 0x8f, 0xb2, 0x6d, 0xdf, 0x1c, 0x19, 0x93, 0x0e, 0x86, 0xba, 0xb4, 0x60, 0x5b,
-	0xfb, 0x97, 0x01, 0x70, 0x98, 0x45, 0x4b, 0xe8, 0x46, 0x94, 0xac, 0xbd, 0x50, 0xf2, 0x5c, 0x78,
-	0x99, 0xa0, 0x41, 0xbd, 0x72, 0x78, 0xb9, 0xf2, 0x23, 0x25, 0xeb, 0x65, 0xa9, 0x5b, 0x09, 0x1a,
-	0xe0, 0xfb, 0xe8, 0x38, 0x45, 0x9f, 0xe1, 0xc9, 0x8e, 0xcb, 0x0d, 0x95, 0xc7, 0x56, 0xe6, 0xa8,
-	0x35, 0xb1, 0xa6, 0xe3, 0x4b, 0xab, 0x6f, 0x5a, 0x7a, 0x30, 0xeb, 0xee, 0x4e, 0x0b, 0xf6, 0x6f,
-	0x03, 0xee, 0x4f, 0xf6, 0xa1, 0x1f, 0xd0, 0x2b, 0xc1, 0x64, 0x8a, 0x48, 0xe5, 0x09, 0x22, 0x49,
-	0x9a, 0xf5, 0x0d, 0xed, 0x3f, 0xbb, 0xf1, 0xd4, 0x92, 0xd5, 0xaa, 0x1c, 0xfb, 0xaa, 0xa7, 0x16,
-	0x4c, 0xc9, 0x02, 0x3f, 0x96, 0x27, 0xc5, 0xc1, 0x3b, 0x78, 0x7a, 0x45, 0x86, 0x7a, 0xd0, 0xda,
-	0xd0, 0x42, 0x33, 0xe9, 0xe0, 0x32, 0x44, 0xcf, 0xa0, 0xbd, 0x25, 0x49, 0x4e, 0x6b, 0xb6, 0x55,
-	0xf2, 0xc6, 0x7c, 0x6d, 0xd8, 0x7f, 0x4c, 0xe8, 0x9e, 0x7d, 0x18, 0x7a, 0x01, 0x50, 0xf1, 0x60,
-	0x24, 0xa5, 0xb5, 0x4d, 0x47, 0x57, 0xbe, 0x90, 0x94, 0xa2, 0x01, 0x3c, 0x94, 0x54, 0x24, 0x71,
-	0x40, 0x32, 0xed, 0xd7, 0xc6, 0x4d, 0x8e, 0xc6, 0xf0, 0x28, 0x8d, 0x99, 0xd7, 0xf4, 0x5b, 0xba,
-	0x6f, 0xa5, 0x31, 0xc3, 0xc7, 0x12, 0xb2, 0x3f, 0x48, 0x1e, 0xd4, 0x12, 0xb2, 0x6f, 0x24, 0xde,
-	0x15, 0x6c, 0x6d, 0x8d, 0xed, 0xd5, 0xcd, 0xdf, 0xf2, 0x9f, 0xc0, 0xbd, 0x7f, 0xfb, 0x7d, 0x1e,
-	0xc6, 0x2a, 0xca, 0x7d, 0x27, 0xe0, 0xa9, 0xab, 0x5f, 0xc5, 0x65, 0x58, 0x05, 0x6e, 0x73, 0x47,
-	0x21, 0x65, 0xae, 0xf0, 0x5f, 0x86, 0xdc, 0x3d, 0x3f, 0x2d, 0xff, 0x4e, 0xdf, 0xd5, 0xec, 0x6f,
-	0x00, 0x00, 0x00, 0xff, 0xff, 0xb4, 0x14, 0x29, 0x8c, 0x75, 0x03, 0x00, 0x00,
+	// 512 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x54, 0xdd, 0x6a, 0xdb, 0x30,
+	0x14, 0xc6, 0x49, 0x53, 0x9a, 0x93, 0x75, 0x49, 0xb5, 0xc1, 0x4c, 0xb6, 0xd1, 0x34, 0x57, 0xb9,
+	0x69, 0x0c, 0x2d, 0x83, 0xb1, 0x11, 0x46, 0x36, 0xba, 0x8e, 0xc2, 0xc6, 0x50, 0x2e, 0x06, 0x83,
+	0x21, 0x64, 0x5b, 0x71, 0xbc, 0xda, 0x92, 0x91, 0xe4, 0xa4, 0x7e, 0xac, 0xdd, 0xec, 0x45, 0xf6,
+	0x0a, 0x7b, 0x90, 0x61, 0x59, 0x75, 0x7e, 0x5a, 0xe8, 0xdd, 0xee, 0xa4, 0xf3, 0xfd, 0x70, 0xce,
+	0x77, 0x6c, 0x41, 0x7f, 0x9e, 0x14, 0x9a, 0xc5, 0x61, 0xe2, 0x65, 0x49, 0x1e, 0xc5, 0x5c, 0x79,
+	0x92, 0x16, 0xe3, 0x4c, 0x0a, 0x2d, 0x50, 0xef, 0x16, 0x1b, 0x5b, 0x6c, 0xf8, 0xd7, 0x81, 0x7d,
+	0x4c, 0x8b, 0x2b, 0xe1, 0xa3, 0x09, 0x74, 0x24, 0x2d, 0x48, 0x90, 0xe4, 0x4a, 0x33, 0xe9, 0x3a,
+	0x03, 0x67, 0xd4, 0x39, 0x7b, 0x31, 0xde, 0x95, 0x8c, 0x31, 0x2d, 0x3e, 0x54, 0x1c, 0x0c, 0xb2,
+	0x3e, 0xa3, 0x63, 0xe8, 0xc8, 0x9c, 0xeb, 0x38, 0x65, 0x84, 0xf1, 0xa5, 0xdb, 0x18, 0x38, 0xa3,
+	0x36, 0x06, 0x5b, 0xba, 0xe0, 0x4b, 0x34, 0x81, 0xe7, 0x6a, 0x91, 0xeb, 0x50, 0xac, 0x38, 0xa1,
+	0x73, 0xcd, 0x24, 0xf9, 0x29, 0x7c, 0x32, 0x8f, 0x79, 0xac, 0x16, 0x4c, 0xb9, 0xcd, 0x81, 0x33,
+	0x3a, 0xc0, 0xee, 0x2d, 0x65, 0x5a, 0x32, 0xae, 0x84, 0xff, 0xd1, 0xe2, 0xe8, 0x2d, 0xf4, 0xb5,
+	0x4e, 0x88, 0x62, 0x81, 0xe0, 0xa1, 0xb2, 0x0e, 0x56, 0x1d, 0xba, 0x7b, 0x03, 0x67, 0xd4, 0xc2,
+	0xcf, 0xb4, 0x4e, 0x66, 0x15, 0xc1, 0xe8, 0xad, 0x38, 0x1c, 0xfe, 0x71, 0x00, 0xd6, 0x7d, 0xa3,
+	0x4b, 0xe8, 0x2e, 0x18, 0x0d, 0x49, 0x24, 0x45, 0x9e, 0x11, 0x95, 0xb1, 0xc0, 0x8e, 0x7b, 0x7c,
+	0x77, 0xdc, 0x4f, 0x8c, 0x86, 0x97, 0x25, 0x6f, 0x96, 0xb1, 0x00, 0x1f, 0x2e, 0x36, 0xaf, 0xe8,
+	0x33, 0x1c, 0xad, 0x84, 0xbc, 0x66, 0x72, 0xd3, 0xaa, 0x31, 0x68, 0x8e, 0x3a, 0x67, 0x27, 0x77,
+	0xad, 0xbe, 0x19, 0xea, 0xda, 0xac, 0xbb, 0xda, 0x2e, 0xa0, 0x53, 0x40, 0x8c, 0x53, 0x3f, 0x61,
+	0x84, 0xe6, 0x5a, 0xa8, 0x80, 0x26, 0x31, 0x8f, 0x6c, 0x32, 0x47, 0x15, 0x32, 0x5d, 0x03, 0xc3,
+	0x5f, 0x0e, 0x1c, 0x6e, 0xb5, 0x87, 0x7e, 0x40, 0xaf, 0xdc, 0xa1, 0xd2, 0x54, 0x6a, 0x92, 0x51,
+	0x49, 0x53, 0xe5, 0x3a, 0xa6, 0x9d, 0xf3, 0x07, 0x26, 0x2b, 0xd7, 0x3a, 0x2b, 0x65, 0x5f, 0x8d,
+	0xea, 0x82, 0x6b, 0x59, 0xe0, 0xc7, 0x72, 0xab, 0xd8, 0x9f, 0xc2, 0x93, 0x7b, 0x68, 0xa8, 0x07,
+	0xcd, 0x6b, 0x56, 0x98, 0x08, 0xdb, 0xb8, 0x3c, 0xa2, 0xa7, 0xd0, 0x5a, 0xd2, 0x24, 0x67, 0xf6,
+	0x33, 0xa8, 0x2e, 0x6f, 0x1a, 0xaf, 0x9d, 0xe1, 0xef, 0x06, 0x74, 0x77, 0x72, 0x40, 0x2f, 0x01,
+	0xaa, 0xf8, 0x38, 0x4d, 0x99, 0xb5, 0x69, 0x9b, 0xca, 0x17, 0x9a, 0x32, 0xd4, 0x87, 0x03, 0xc9,
+	0xb2, 0x24, 0x0e, 0xa8, 0x32, 0x7e, 0x2d, 0x5c, 0xdf, 0xd1, 0x09, 0x3c, 0x4a, 0x63, 0x4e, 0x6a,
+	0xbc, 0x69, 0xf0, 0x4e, 0x1a, 0x73, 0xbc, 0x49, 0xa1, 0x37, 0x6b, 0xca, 0x9e, 0xa5, 0xd0, 0x9b,
+	0x9a, 0x42, 0xee, 0x89, 0xad, 0x65, 0x62, 0x7b, 0xf5, 0xe0, 0x16, 0xff, 0x53, 0x70, 0xef, 0xdf,
+	0x7d, 0x9f, 0x44, 0xb1, 0x5e, 0xe4, 0xfe, 0x38, 0x10, 0xa9, 0x67, 0xba, 0x12, 0x32, 0xaa, 0x0e,
+	0x5e, 0xfd, 0xcf, 0x47, 0x8c, 0x7b, 0x99, 0x7f, 0x1a, 0x09, 0x6f, 0xf7, 0x19, 0xf0, 0xf7, 0xcd,
+	0x1b, 0x70, 0xfe, 0x2f, 0x00, 0x00, 0xff, 0xff, 0xe5, 0x00, 0x39, 0x5b, 0x21, 0x04, 0x00, 0x00,
 }
diff --git a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.validate.go b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.validate.go
index e85e27a8ac..b079e610f5 100644
--- a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.validate.go
+++ b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.validate.go
@@ -55,6 +55,10 @@ func (m *RayJob) Validate() error {
 
 	// no validation rules for RuntimeEnv
 
+	// no validation rules for ShutdownAfterJobFinishes
+
+	// no validation rules for TtlSecondsAfterFinished
+
 	return nil
 }
 
@@ -144,6 +148,8 @@ func (m *RayCluster) Validate() error {
 
 	}
 
+	// no validation rules for EnableAutoscaling
+
 	return nil
 }
 
diff --git a/flyteidl/gen/pb-java/flyteidl/plugins/Ray.java b/flyteidl/gen/pb-java/flyteidl/plugins/Ray.java
index f035e962e8..5dd419f827 100644
--- a/flyteidl/gen/pb-java/flyteidl/plugins/Ray.java
+++ b/flyteidl/gen/pb-java/flyteidl/plugins/Ray.java
@@ -62,6 +62,24 @@ public interface RayJobOrBuilder extends
      */
     com.google.protobuf.ByteString
         getRuntimeEnvBytes();
+
+    /**
+     * <pre>
+     * shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+     * </pre>
+     *
+     * <code>bool shutdown_after_job_finishes = 3;</code>
+     */
+    boolean getShutdownAfterJobFinishes();
+
+    /**
+     * <pre>
+     * ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+     * </pre>
+     *
+     * <code>int32 ttl_seconds_after_finished = 4;</code>
+     */
+    int getTtlSecondsAfterFinished();
   }
   /**
    * <pre>
@@ -126,6 +144,16 @@ private RayJob(
               runtimeEnv_ = s;
               break;
             }
+            case 24: {
+
+              shutdownAfterJobFinishes_ = input.readBool();
+              break;
+            }
+            case 32: {
+
+              ttlSecondsAfterFinished_ = input.readInt32();
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -235,6 +263,32 @@ public java.lang.String getRuntimeEnv() {
       }
     }
 
+    public static final int SHUTDOWN_AFTER_JOB_FINISHES_FIELD_NUMBER = 3;
+    private boolean shutdownAfterJobFinishes_;
+    /**
+     * <pre>
+     * shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+     * </pre>
+     *
+     * <code>bool shutdown_after_job_finishes = 3;</code>
+     */
+    public boolean getShutdownAfterJobFinishes() {
+      return shutdownAfterJobFinishes_;
+    }
+
+    public static final int TTL_SECONDS_AFTER_FINISHED_FIELD_NUMBER = 4;
+    private int ttlSecondsAfterFinished_;
+    /**
+     * <pre>
+     * ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+     * </pre>
+     *
+     * <code>int32 ttl_seconds_after_finished = 4;</code>
+     */
+    public int getTtlSecondsAfterFinished() {
+      return ttlSecondsAfterFinished_;
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -255,6 +309,12 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       if (!getRuntimeEnvBytes().isEmpty()) {
         com.google.protobuf.GeneratedMessageV3.writeString(output, 2, runtimeEnv_);
       }
+      if (shutdownAfterJobFinishes_ != false) {
+        output.writeBool(3, shutdownAfterJobFinishes_);
+      }
+      if (ttlSecondsAfterFinished_ != 0) {
+        output.writeInt32(4, ttlSecondsAfterFinished_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -271,6 +331,14 @@ public int getSerializedSize() {
       if (!getRuntimeEnvBytes().isEmpty()) {
         size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, runtimeEnv_);
       }
+      if (shutdownAfterJobFinishes_ != false) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeBoolSize(3, shutdownAfterJobFinishes_);
+      }
+      if (ttlSecondsAfterFinished_ != 0) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeInt32Size(4, ttlSecondsAfterFinished_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -293,6 +361,10 @@ public boolean equals(final java.lang.Object obj) {
       }
       if (!getRuntimeEnv()
           .equals(other.getRuntimeEnv())) return false;
+      if (getShutdownAfterJobFinishes()
+          != other.getShutdownAfterJobFinishes()) return false;
+      if (getTtlSecondsAfterFinished()
+          != other.getTtlSecondsAfterFinished()) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -310,6 +382,11 @@ public int hashCode() {
       }
       hash = (37 * hash) + RUNTIME_ENV_FIELD_NUMBER;
       hash = (53 * hash) + getRuntimeEnv().hashCode();
+      hash = (37 * hash) + SHUTDOWN_AFTER_JOB_FINISHES_FIELD_NUMBER;
+      hash = (53 * hash) + com.google.protobuf.Internal.hashBoolean(
+          getShutdownAfterJobFinishes());
+      hash = (37 * hash) + TTL_SECONDS_AFTER_FINISHED_FIELD_NUMBER;
+      hash = (53 * hash) + getTtlSecondsAfterFinished();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -455,6 +532,10 @@ public Builder clear() {
         }
         runtimeEnv_ = "";
 
+        shutdownAfterJobFinishes_ = false;
+
+        ttlSecondsAfterFinished_ = 0;
+
         return this;
       }
 
@@ -487,6 +568,8 @@ public flyteidl.plugins.Ray.RayJob buildPartial() {
           result.rayCluster_ = rayClusterBuilder_.build();
         }
         result.runtimeEnv_ = runtimeEnv_;
+        result.shutdownAfterJobFinishes_ = shutdownAfterJobFinishes_;
+        result.ttlSecondsAfterFinished_ = ttlSecondsAfterFinished_;
         onBuilt();
         return result;
       }
@@ -542,6 +625,12 @@ public Builder mergeFrom(flyteidl.plugins.Ray.RayJob other) {
           runtimeEnv_ = other.runtimeEnv_;
           onChanged();
         }
+        if (other.getShutdownAfterJobFinishes() != false) {
+          setShutdownAfterJobFinishes(other.getShutdownAfterJobFinishes());
+        }
+        if (other.getTtlSecondsAfterFinished() != 0) {
+          setTtlSecondsAfterFinished(other.getTtlSecondsAfterFinished());
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -817,6 +906,82 @@ public Builder setRuntimeEnvBytes(
         onChanged();
         return this;
       }
+
+      private boolean shutdownAfterJobFinishes_ ;
+      /**
+       * <pre>
+       * shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>bool shutdown_after_job_finishes = 3;</code>
+       */
+      public boolean getShutdownAfterJobFinishes() {
+        return shutdownAfterJobFinishes_;
+      }
+      /**
+       * <pre>
+       * shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>bool shutdown_after_job_finishes = 3;</code>
+       */
+      public Builder setShutdownAfterJobFinishes(boolean value) {
+        
+        shutdownAfterJobFinishes_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>bool shutdown_after_job_finishes = 3;</code>
+       */
+      public Builder clearShutdownAfterJobFinishes() {
+        
+        shutdownAfterJobFinishes_ = false;
+        onChanged();
+        return this;
+      }
+
+      private int ttlSecondsAfterFinished_ ;
+      /**
+       * <pre>
+       * ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>int32 ttl_seconds_after_finished = 4;</code>
+       */
+      public int getTtlSecondsAfterFinished() {
+        return ttlSecondsAfterFinished_;
+      }
+      /**
+       * <pre>
+       * ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>int32 ttl_seconds_after_finished = 4;</code>
+       */
+      public Builder setTtlSecondsAfterFinished(int value) {
+        
+        ttlSecondsAfterFinished_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+       * </pre>
+       *
+       * <code>int32 ttl_seconds_after_finished = 4;</code>
+       */
+      public Builder clearTtlSecondsAfterFinished() {
+        
+        ttlSecondsAfterFinished_ = 0;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -942,6 +1107,15 @@ public interface RayClusterOrBuilder extends
      */
     flyteidl.plugins.Ray.WorkerGroupSpecOrBuilder getWorkerGroupSpecOrBuilder(
         int index);
+
+    /**
+     * <pre>
+     * Whether to enable autoscaling.
+     * </pre>
+     *
+     * <code>bool enable_autoscaling = 3;</code>
+     */
+    boolean getEnableAutoscaling();
   }
   /**
    * <pre>
@@ -1009,6 +1183,11 @@ private RayCluster(
                   input.readMessage(flyteidl.plugins.Ray.WorkerGroupSpec.parser(), extensionRegistry));
               break;
             }
+            case 24: {
+
+              enableAutoscaling_ = input.readBool();
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -1133,6 +1312,19 @@ public flyteidl.plugins.Ray.WorkerGroupSpecOrBuilder getWorkerGroupSpecOrBuilder
       return workerGroupSpec_.get(index);
     }
 
+    public static final int ENABLE_AUTOSCALING_FIELD_NUMBER = 3;
+    private boolean enableAutoscaling_;
+    /**
+     * <pre>
+     * Whether to enable autoscaling.
+     * </pre>
+     *
+     * <code>bool enable_autoscaling = 3;</code>
+     */
+    public boolean getEnableAutoscaling() {
+      return enableAutoscaling_;
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -1153,6 +1345,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       for (int i = 0; i < workerGroupSpec_.size(); i++) {
         output.writeMessage(2, workerGroupSpec_.get(i));
       }
+      if (enableAutoscaling_ != false) {
+        output.writeBool(3, enableAutoscaling_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -1170,6 +1365,10 @@ public int getSerializedSize() {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(2, workerGroupSpec_.get(i));
       }
+      if (enableAutoscaling_ != false) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeBoolSize(3, enableAutoscaling_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -1192,6 +1391,8 @@ public boolean equals(final java.lang.Object obj) {
       }
       if (!getWorkerGroupSpecList()
           .equals(other.getWorkerGroupSpecList())) return false;
+      if (getEnableAutoscaling()
+          != other.getEnableAutoscaling()) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -1211,6 +1412,9 @@ public int hashCode() {
         hash = (37 * hash) + WORKER_GROUP_SPEC_FIELD_NUMBER;
         hash = (53 * hash) + getWorkerGroupSpecList().hashCode();
       }
+      hash = (37 * hash) + ENABLE_AUTOSCALING_FIELD_NUMBER;
+      hash = (53 * hash) + com.google.protobuf.Internal.hashBoolean(
+          getEnableAutoscaling());
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -1361,6 +1565,8 @@ public Builder clear() {
         } else {
           workerGroupSpecBuilder_.clear();
         }
+        enableAutoscaling_ = false;
+
         return this;
       }
 
@@ -1403,6 +1609,7 @@ public flyteidl.plugins.Ray.RayCluster buildPartial() {
         } else {
           result.workerGroupSpec_ = workerGroupSpecBuilder_.build();
         }
+        result.enableAutoscaling_ = enableAutoscaling_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -1481,6 +1688,9 @@ public Builder mergeFrom(flyteidl.plugins.Ray.RayCluster other) {
             }
           }
         }
+        if (other.getEnableAutoscaling() != false) {
+          setEnableAutoscaling(other.getEnableAutoscaling());
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -1975,6 +2185,44 @@ public flyteidl.plugins.Ray.WorkerGroupSpec.Builder addWorkerGroupSpecBuilder(
         }
         return workerGroupSpecBuilder_;
       }
+
+      private boolean enableAutoscaling_ ;
+      /**
+       * <pre>
+       * Whether to enable autoscaling.
+       * </pre>
+       *
+       * <code>bool enable_autoscaling = 3;</code>
+       */
+      public boolean getEnableAutoscaling() {
+        return enableAutoscaling_;
+      }
+      /**
+       * <pre>
+       * Whether to enable autoscaling.
+       * </pre>
+       *
+       * <code>bool enable_autoscaling = 3;</code>
+       */
+      public Builder setEnableAutoscaling(boolean value) {
+        
+        enableAutoscaling_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * Whether to enable autoscaling.
+       * </pre>
+       *
+       * <code>bool enable_autoscaling = 3;</code>
+       */
+      public Builder clearEnableAutoscaling() {
+        
+        enableAutoscaling_ = false;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -4108,24 +4356,26 @@ public flyteidl.plugins.Ray.WorkerGroupSpec getDefaultInstanceForType() {
   static {
     java.lang.String[] descriptorData = {
       "\n\032flyteidl/plugins/ray.proto\022\020flyteidl.p" +
-      "lugins\"P\n\006RayJob\0221\n\013ray_cluster\030\001 \001(\0132\034." +
-      "flyteidl.plugins.RayCluster\022\023\n\013runtime_e" +
-      "nv\030\002 \001(\t\"\204\001\n\nRayCluster\0228\n\017head_group_sp" +
-      "ec\030\001 \001(\0132\037.flyteidl.plugins.HeadGroupSpe" +
-      "c\022<\n\021worker_group_spec\030\002 \003(\0132!.flyteidl." +
-      "plugins.WorkerGroupSpec\"\225\001\n\rHeadGroupSpe" +
-      "c\022M\n\020ray_start_params\030\001 \003(\01323.flyteidl.p" +
-      "lugins.HeadGroupSpec.RayStartParamsEntry" +
-      "\0325\n\023RayStartParamsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005" +
-      "value\030\002 \001(\t:\0028\001\"\353\001\n\017WorkerGroupSpec\022\022\n\ng" +
-      "roup_name\030\001 \001(\t\022\020\n\010replicas\030\002 \001(\005\022\024\n\014min" +
-      "_replicas\030\003 \001(\005\022\024\n\014max_replicas\030\004 \001(\005\022O\n" +
-      "\020ray_start_params\030\005 \003(\01325.flyteidl.plugi" +
-      "ns.WorkerGroupSpec.RayStartParamsEntry\0325" +
-      "\n\023RayStartParamsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005va" +
-      "lue\030\002 \001(\t:\0028\001B?Z=github.com/flyteorg/fly" +
-      "te/flyteidl/gen/pb-go/flyteidl/pluginsb\006" +
-      "proto3"
+      "lugins\"\231\001\n\006RayJob\0221\n\013ray_cluster\030\001 \001(\0132\034" +
+      ".flyteidl.plugins.RayCluster\022\023\n\013runtime_" +
+      "env\030\002 \001(\t\022#\n\033shutdown_after_job_finishes" +
+      "\030\003 \001(\010\022\"\n\032ttl_seconds_after_finished\030\004 \001" +
+      "(\005\"\240\001\n\nRayCluster\0228\n\017head_group_spec\030\001 \001" +
+      "(\0132\037.flyteidl.plugins.HeadGroupSpec\022<\n\021w" +
+      "orker_group_spec\030\002 \003(\0132!.flyteidl.plugin" +
+      "s.WorkerGroupSpec\022\032\n\022enable_autoscaling\030" +
+      "\003 \001(\010\"\225\001\n\rHeadGroupSpec\022M\n\020ray_start_par" +
+      "ams\030\001 \003(\01323.flyteidl.plugins.HeadGroupSp" +
+      "ec.RayStartParamsEntry\0325\n\023RayStartParams" +
+      "Entry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\353\001" +
+      "\n\017WorkerGroupSpec\022\022\n\ngroup_name\030\001 \001(\t\022\020\n" +
+      "\010replicas\030\002 \001(\005\022\024\n\014min_replicas\030\003 \001(\005\022\024\n" +
+      "\014max_replicas\030\004 \001(\005\022O\n\020ray_start_params\030" +
+      "\005 \003(\01325.flyteidl.plugins.WorkerGroupSpec" +
+      ".RayStartParamsEntry\0325\n\023RayStartParamsEn" +
+      "try\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001B?Z=g" +
+      "ithub.com/flyteorg/flyte/flyteidl/gen/pb" +
+      "-go/flyteidl/pluginsb\006proto3"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
         new com.google.protobuf.Descriptors.FileDescriptor.    InternalDescriptorAssigner() {
@@ -4144,13 +4394,13 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors(
     internal_static_flyteidl_plugins_RayJob_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_flyteidl_plugins_RayJob_descriptor,
-        new java.lang.String[] { "RayCluster", "RuntimeEnv", });
+        new java.lang.String[] { "RayCluster", "RuntimeEnv", "ShutdownAfterJobFinishes", "TtlSecondsAfterFinished", });
     internal_static_flyteidl_plugins_RayCluster_descriptor =
       getDescriptor().getMessageTypes().get(1);
     internal_static_flyteidl_plugins_RayCluster_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_flyteidl_plugins_RayCluster_descriptor,
-        new java.lang.String[] { "HeadGroupSpec", "WorkerGroupSpec", });
+        new java.lang.String[] { "HeadGroupSpec", "WorkerGroupSpec", "EnableAutoscaling", });
     internal_static_flyteidl_plugins_HeadGroupSpec_descriptor =
       getDescriptor().getMessageTypes().get(2);
     internal_static_flyteidl_plugins_HeadGroupSpec_fieldAccessorTable = new
diff --git a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
index 08c33c782c..1c5de3b666 100644
--- a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
+++ b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
@@ -13,7 +13,7 @@
 
 
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/plugins/ray.proto\x12\x10\x66lyteidl.plugins\"h\n\x06RayJob\x12=\n\x0bray_cluster\x18\x01 \x01(\x0b\x32\x1c.flyteidl.plugins.RayClusterR\nrayCluster\x12\x1f\n\x0bruntime_env\x18\x02 \x01(\tR\nruntimeEnv\"\xa4\x01\n\nRayCluster\x12G\n\x0fhead_group_spec\x18\x01 \x01(\x0b\x32\x1f.flyteidl.plugins.HeadGroupSpecR\rheadGroupSpec\x12M\n\x11worker_group_spec\x18\x02 \x03(\x0b\x32!.flyteidl.plugins.WorkerGroupSpecR\x0fworkerGroupSpec\"\xb1\x01\n\rHeadGroupSpec\x12]\n\x10ray_start_params\x18\x01 \x03(\x0b\x32\x33.flyteidl.plugins.HeadGroupSpec.RayStartParamsEntryR\x0erayStartParams\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xb6\x02\n\x0fWorkerGroupSpec\x12\x1d\n\ngroup_name\x18\x01 \x01(\tR\tgroupName\x12\x1a\n\x08replicas\x18\x02 \x01(\x05R\x08replicas\x12!\n\x0cmin_replicas\x18\x03 \x01(\x05R\x0bminReplicas\x12!\n\x0cmax_replicas\x18\x04 \x01(\x05R\x0bmaxReplicas\x12_\n\x10ray_start_params\x18\x05 \x03(\x0b\x32\x35.flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntryR\x0erayStartParams\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\xc0\x01\n\x14\x63om.flyteidl.pluginsB\x08RayProtoP\x01Z=github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/plugins\xa2\x02\x03\x46PX\xaa\x02\x10\x46lyteidl.Plugins\xca\x02\x10\x46lyteidl\\Plugins\xe2\x02\x1c\x46lyteidl\\Plugins\\GPBMetadata\xea\x02\x11\x46lyteidl::Pluginsb\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/plugins/ray.proto\x12\x10\x66lyteidl.plugins\"\xe4\x01\n\x06RayJob\x12=\n\x0bray_cluster\x18\x01 \x01(\x0b\x32\x1c.flyteidl.plugins.RayClusterR\nrayCluster\x12\x1f\n\x0bruntime_env\x18\x02 \x01(\tR\nruntimeEnv\x12=\n\x1bshutdown_after_job_finishes\x18\x03 \x01(\x08R\x18shutdownAfterJobFinishes\x12;\n\x1attl_seconds_after_finished\x18\x04 \x01(\x05R\x17ttlSecondsAfterFinished\"\xd3\x01\n\nRayCluster\x12G\n\x0fhead_group_spec\x18\x01 \x01(\x0b\x32\x1f.flyteidl.plugins.HeadGroupSpecR\rheadGroupSpec\x12M\n\x11worker_group_spec\x18\x02 \x03(\x0b\x32!.flyteidl.plugins.WorkerGroupSpecR\x0fworkerGroupSpec\x12-\n\x12\x65nable_autoscaling\x18\x03 \x01(\x08R\x11\x65nableAutoscaling\"\xb1\x01\n\rHeadGroupSpec\x12]\n\x10ray_start_params\x18\x01 \x03(\x0b\x32\x33.flyteidl.plugins.HeadGroupSpec.RayStartParamsEntryR\x0erayStartParams\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xb6\x02\n\x0fWorkerGroupSpec\x12\x1d\n\ngroup_name\x18\x01 \x01(\tR\tgroupName\x12\x1a\n\x08replicas\x18\x02 \x01(\x05R\x08replicas\x12!\n\x0cmin_replicas\x18\x03 \x01(\x05R\x0bminReplicas\x12!\n\x0cmax_replicas\x18\x04 \x01(\x05R\x0bmaxReplicas\x12_\n\x10ray_start_params\x18\x05 \x03(\x0b\x32\x35.flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntryR\x0erayStartParams\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\xc0\x01\n\x14\x63om.flyteidl.pluginsB\x08RayProtoP\x01Z=github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/plugins\xa2\x02\x03\x46PX\xaa\x02\x10\x46lyteidl.Plugins\xca\x02\x10\x46lyteidl\\Plugins\xe2\x02\x1c\x46lyteidl\\Plugins\\GPBMetadata\xea\x02\x11\x46lyteidl::Pluginsb\x06proto3')
 
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
@@ -26,16 +26,16 @@
   _HEADGROUPSPEC_RAYSTARTPARAMSENTRY._serialized_options = b'8\001'
   _WORKERGROUPSPEC_RAYSTARTPARAMSENTRY._options = None
   _WORKERGROUPSPEC_RAYSTARTPARAMSENTRY._serialized_options = b'8\001'
-  _globals['_RAYJOB']._serialized_start=48
-  _globals['_RAYJOB']._serialized_end=152
-  _globals['_RAYCLUSTER']._serialized_start=155
-  _globals['_RAYCLUSTER']._serialized_end=319
-  _globals['_HEADGROUPSPEC']._serialized_start=322
-  _globals['_HEADGROUPSPEC']._serialized_end=499
-  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=434
-  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=499
-  _globals['_WORKERGROUPSPEC']._serialized_start=502
-  _globals['_WORKERGROUPSPEC']._serialized_end=812
-  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=434
-  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=499
+  _globals['_RAYJOB']._serialized_start=49
+  _globals['_RAYJOB']._serialized_end=277
+  _globals['_RAYCLUSTER']._serialized_start=280
+  _globals['_RAYCLUSTER']._serialized_end=491
+  _globals['_HEADGROUPSPEC']._serialized_start=494
+  _globals['_HEADGROUPSPEC']._serialized_end=671
+  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=606
+  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=671
+  _globals['_WORKERGROUPSPEC']._serialized_start=674
+  _globals['_WORKERGROUPSPEC']._serialized_end=984
+  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=606
+  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=671
 # @@protoc_insertion_point(module_scope)
diff --git a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
index 16b67e478a..15c78912bd 100644
--- a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
+++ b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
@@ -6,20 +6,26 @@ from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Map
 DESCRIPTOR: _descriptor.FileDescriptor
 
 class RayJob(_message.Message):
-    __slots__ = ["ray_cluster", "runtime_env"]
+    __slots__ = ["ray_cluster", "runtime_env", "shutdown_after_job_finishes", "ttl_seconds_after_finished"]
     RAY_CLUSTER_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_ENV_FIELD_NUMBER: _ClassVar[int]
+    SHUTDOWN_AFTER_JOB_FINISHES_FIELD_NUMBER: _ClassVar[int]
+    TTL_SECONDS_AFTER_FINISHED_FIELD_NUMBER: _ClassVar[int]
     ray_cluster: RayCluster
     runtime_env: str
-    def __init__(self, ray_cluster: _Optional[_Union[RayCluster, _Mapping]] = ..., runtime_env: _Optional[str] = ...) -> None: ...
+    shutdown_after_job_finishes: bool
+    ttl_seconds_after_finished: int
+    def __init__(self, ray_cluster: _Optional[_Union[RayCluster, _Mapping]] = ..., runtime_env: _Optional[str] = ..., shutdown_after_job_finishes: bool = ..., ttl_seconds_after_finished: _Optional[int] = ...) -> None: ...
 
 class RayCluster(_message.Message):
-    __slots__ = ["head_group_spec", "worker_group_spec"]
+    __slots__ = ["head_group_spec", "worker_group_spec", "enable_autoscaling"]
     HEAD_GROUP_SPEC_FIELD_NUMBER: _ClassVar[int]
     WORKER_GROUP_SPEC_FIELD_NUMBER: _ClassVar[int]
+    ENABLE_AUTOSCALING_FIELD_NUMBER: _ClassVar[int]
     head_group_spec: HeadGroupSpec
     worker_group_spec: _containers.RepeatedCompositeFieldContainer[WorkerGroupSpec]
-    def __init__(self, head_group_spec: _Optional[_Union[HeadGroupSpec, _Mapping]] = ..., worker_group_spec: _Optional[_Iterable[_Union[WorkerGroupSpec, _Mapping]]] = ...) -> None: ...
+    enable_autoscaling: bool
+    def __init__(self, head_group_spec: _Optional[_Union[HeadGroupSpec, _Mapping]] = ..., worker_group_spec: _Optional[_Iterable[_Union[WorkerGroupSpec, _Mapping]]] = ..., enable_autoscaling: bool = ...) -> None: ...
 
 class HeadGroupSpec(_message.Message):
     __slots__ = ["ray_start_params"]
diff --git a/flyteidl/gen/pb_rust/flyteidl.plugins.rs b/flyteidl/gen/pb_rust/flyteidl.plugins.rs
index 11e4ad05af..bfed40f1a8 100644
--- a/flyteidl/gen/pb_rust/flyteidl.plugins.rs
+++ b/flyteidl/gen/pb_rust/flyteidl.plugins.rs
@@ -175,6 +175,12 @@ pub struct RayJob {
     /// Ray runtime environments: <https://docs.ray.io/en/latest/ray-core/handling-dependencies.html#runtime-environments>
     #[prost(string, tag="2")]
     pub runtime_env: ::prost::alloc::string::String,
+    /// shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+    #[prost(bool, tag="3")]
+    pub shutdown_after_job_finishes: bool,
+    /// ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+    #[prost(int32, tag="4")]
+    pub ttl_seconds_after_finished: i32,
 }
 /// Define Ray cluster defines the desired state of RayCluster
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -186,6 +192,9 @@ pub struct RayCluster {
     /// WorkerGroupSpecs are the specs for the worker pods
     #[prost(message, repeated, tag="2")]
     pub worker_group_spec: ::prost::alloc::vec::Vec<WorkerGroupSpec>,
+    /// Whether to enable autoscaling.
+    #[prost(bool, tag="3")]
+    pub enable_autoscaling: bool,
 }
 /// HeadGroupSpec are the spec for the head pod
 #[allow(clippy::derive_partial_eq_without_eq)]
diff --git a/flyteidl/protos/flyteidl/plugins/ray.proto b/flyteidl/protos/flyteidl/plugins/ray.proto
index 1f80a4b4f3..1afcee8d93 100644
--- a/flyteidl/protos/flyteidl/plugins/ray.proto
+++ b/flyteidl/protos/flyteidl/plugins/ray.proto
@@ -11,6 +11,10 @@ message RayJob {
   // runtime_env is base64 encoded.
   // Ray runtime environments: https://docs.ray.io/en/latest/ray-core/handling-dependencies.html#runtime-environments
   string runtime_env = 2;
+  // shutdown_after_job_finishes specifies whether the RayCluster should be deleted after the RayJob finishes.
+  bool shutdown_after_job_finishes = 3;
+  // ttl_seconds_after_finished specifies the number of seconds after which the RayCluster will be deleted after the RayJob finishes.
+  int32 ttl_seconds_after_finished = 4;
 }
 
 // Define Ray cluster defines the desired state of RayCluster
@@ -19,6 +23,8 @@ message RayCluster {
   HeadGroupSpec head_group_spec = 1;
   // WorkerGroupSpecs are the specs for the worker pods
   repeated WorkerGroupSpec worker_group_spec = 2;
+  // Whether to enable autoscaling.
+  bool enable_autoscaling = 3;
 }
 
 // HeadGroupSpec are the spec for the head pod
diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
index 50a9c76094..f6abb58c93 100644
--- a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
+++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
@@ -46,8 +46,7 @@ var logTemplateRegexes = struct {
 	tasklog.MustCreateRegex("rayJobID"),
 }
 
-type rayJobResourceHandler struct {
-}
+type rayJobResourceHandler struct{}
 
 func (rayJobResourceHandler) GetProperties() k8s.PluginProperties {
 	return k8s.PluginProperties{}
@@ -128,7 +127,8 @@ func (rayJobResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsC
 			EnableIngress:  &enableIngress,
 			RayStartParams: headNodeRayStartParams,
 		},
-		WorkerGroupSpecs: []rayv1alpha1.WorkerGroupSpec{},
+		WorkerGroupSpecs:        []rayv1alpha1.WorkerGroupSpec{},
+		EnableInTreeAutoscaling: &rayJob.RayCluster.EnableAutoscaling,
 	}
 
 	for _, spec := range rayJob.RayCluster.WorkerGroupSpec {
@@ -140,16 +140,6 @@ func (rayJobResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsC
 			taskCtx,
 		)
 
-		minReplicas := spec.Replicas
-		maxReplicas := spec.Replicas
-		if spec.MinReplicas != 0 {
-			minReplicas = spec.MinReplicas
-		}
-
-		if spec.MaxReplicas != 0 {
-			maxReplicas = spec.MaxReplicas
-		}
-
 		workerNodeRayStartParams := make(map[string]string)
 		if spec.RayStartParams != nil {
 			workerNodeRayStartParams = spec.RayStartParams
@@ -165,6 +155,15 @@ func (rayJobResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsC
 			workerNodeRayStartParams[DisableUsageStatsStartParameter] = "true"
 		}
 
+		minReplicas := spec.MinReplicas
+		if minReplicas > spec.Replicas {
+			minReplicas = spec.Replicas
+		}
+		maxReplicas := spec.MaxReplicas
+		if maxReplicas < spec.Replicas {
+			maxReplicas = spec.Replicas
+		}
+
 		workerNodeSpec := rayv1alpha1.WorkerGroupSpec{
 			GroupName:      spec.GroupName,
 			MinReplicas:    &minReplicas,
@@ -184,11 +183,18 @@ func (rayJobResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsC
 		rayClusterSpec.WorkerGroupSpecs[index].Template.Spec.ServiceAccountName = serviceAccountName
 	}
 
+	shutdownAfterJobFinishes := cfg.ShutdownAfterJobFinishes
+	ttlSecondsAfterFinished := &cfg.TTLSecondsAfterFinished
+	if rayJob.ShutdownAfterJobFinishes {
+		shutdownAfterJobFinishes = true
+		ttlSecondsAfterFinished = &rayJob.TtlSecondsAfterFinished
+	}
+
 	jobSpec := rayv1alpha1.RayJobSpec{
 		RayClusterSpec:           rayClusterSpec,
 		Entrypoint:               strings.Join(primaryContainer.Args, " "),
-		ShutdownAfterJobFinishes: cfg.ShutdownAfterJobFinishes,
-		TTLSecondsAfterFinished:  &cfg.TTLSecondsAfterFinished,
+		ShutdownAfterJobFinishes: shutdownAfterJobFinishes,
+		TTLSecondsAfterFinished:  ttlSecondsAfterFinished,
 		RuntimeEnv:               rayJob.RuntimeEnv,
 	}
 
diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
index 64700957c9..b171ae9aa8 100644
--- a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
+++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
@@ -28,8 +28,10 @@ import (
 	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/utils"
 )
 
-const testImage = "image://"
-const serviceAccount = "ray_sa"
+const (
+	testImage      = "image://"
+	serviceAccount = "ray_sa"
+)
 
 var (
 	dummyEnvVars = []*core.KeyValuePair{
@@ -79,9 +81,12 @@ func transformPodSpecToTaskTemplateTarget(podSpec *corev1.PodSpec) *core.TaskTem
 func dummyRayCustomObj() *plugins.RayJob {
 	return &plugins.RayJob{
 		RayCluster: &plugins.RayCluster{
-			HeadGroupSpec:   &plugins.HeadGroupSpec{RayStartParams: map[string]string{"num-cpus": "1"}},
-			WorkerGroupSpec: []*plugins.WorkerGroupSpec{{GroupName: workerGroupName, Replicas: 3}},
+			HeadGroupSpec:     &plugins.HeadGroupSpec{RayStartParams: map[string]string{"num-cpus": "1"}},
+			WorkerGroupSpec:   []*plugins.WorkerGroupSpec{{GroupName: workerGroupName, Replicas: 3, MinReplicas: 3, MaxReplicas: 3}},
+			EnableAutoscaling: true,
 		},
+		ShutdownAfterJobFinishes: true,
+		TtlSecondsAfterFinished:  120,
 	}
 }
 
@@ -176,21 +181,26 @@ func TestBuildResourceRay(t *testing.T) {
 	ray, ok := RayResource.(*rayv1alpha1.RayJob)
 	assert.True(t, ok)
 
+	assert.Equal(t, *ray.Spec.RayClusterSpec.EnableInTreeAutoscaling, true)
+	assert.Equal(t, ray.Spec.ShutdownAfterJobFinishes, true)
+	assert.Equal(t, *ray.Spec.TTLSecondsAfterFinished, int32(120))
+
 	headReplica := int32(1)
-	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Replicas, &headReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.HeadGroupSpec.Replicas, headReplica)
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.ServiceAccountName, serviceAccount)
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.RayStartParams,
 		map[string]string{
 			"dashboard-host": "0.0.0.0", "disable-usage-stats": "true", "include-dashboard": "true",
-			"node-ip-address": "$MY_POD_IP", "num-cpus": "1"})
+			"node-ip-address": "$MY_POD_IP", "num-cpus": "1",
+		})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Annotations, map[string]string{"annotation-1": "val1"})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Labels, map[string]string{"label-1": "val1"})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.Tolerations, toleration)
 
 	workerReplica := int32(3)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Replicas, &workerReplica)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MinReplicas, &workerReplica)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MaxReplicas, &workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Replicas, workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MinReplicas, workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MaxReplicas, workerReplica)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].GroupName, workerGroupName)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Spec.ServiceAccountName, serviceAccount)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].RayStartParams, map[string]string{"disable-usage-stats": "true", "node-ip-address": "$MY_POD_IP"})
@@ -230,7 +240,7 @@ func TestBuildResourceRayExtendedResources(t *testing.T) {
 			[]corev1.NodeSelectorTerm{
 				{
 					MatchExpressions: []corev1.NodeSelectorRequirement{
-						corev1.NodeSelectorRequirement{
+						{
 							Key:      "gpu-node-label",
 							Operator: corev1.NodeSelectorOpIn,
 							Values:   []string{"nvidia-tesla-t4"},
@@ -270,12 +280,12 @@ func TestBuildResourceRayExtendedResources(t *testing.T) {
 			[]corev1.NodeSelectorTerm{
 				{
 					MatchExpressions: []corev1.NodeSelectorRequirement{
-						corev1.NodeSelectorRequirement{
+						{
 							Key:      "gpu-node-label",
 							Operator: corev1.NodeSelectorOpIn,
 							Values:   []string{"nvidia-tesla-a100"},
 						},
-						corev1.NodeSelectorRequirement{
+						{
 							Key:      "gpu-partition-size",
 							Operator: corev1.NodeSelectorOpIn,
 							Values:   []string{"1g.5gb"},
@@ -345,9 +355,12 @@ func TestDefaultStartParameters(t *testing.T) {
 	rayJobResourceHandler := rayJobResourceHandler{}
 	rayJob := &plugins.RayJob{
 		RayCluster: &plugins.RayCluster{
-			HeadGroupSpec:   &plugins.HeadGroupSpec{},
-			WorkerGroupSpec: []*plugins.WorkerGroupSpec{{GroupName: workerGroupName, Replicas: 3}},
+			HeadGroupSpec:     &plugins.HeadGroupSpec{},
+			WorkerGroupSpec:   []*plugins.WorkerGroupSpec{{GroupName: workerGroupName, Replicas: 3, MinReplicas: 3, MaxReplicas: 3}},
+			EnableAutoscaling: true,
 		},
+		ShutdownAfterJobFinishes: true,
+		TtlSecondsAfterFinished:  120,
 	}
 
 	taskTemplate := dummyRayTaskTemplate("ray-id", rayJob)
@@ -367,21 +380,26 @@ func TestDefaultStartParameters(t *testing.T) {
 	ray, ok := RayResource.(*rayv1alpha1.RayJob)
 	assert.True(t, ok)
 
+	assert.Equal(t, *ray.Spec.RayClusterSpec.EnableInTreeAutoscaling, true)
+	assert.Equal(t, ray.Spec.ShutdownAfterJobFinishes, true)
+	assert.Equal(t, *ray.Spec.TTLSecondsAfterFinished, int32(120))
+
 	headReplica := int32(1)
-	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Replicas, &headReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.HeadGroupSpec.Replicas, headReplica)
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.ServiceAccountName, serviceAccount)
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.RayStartParams,
 		map[string]string{
 			"dashboard-host": "0.0.0.0", "disable-usage-stats": "true", "include-dashboard": "true",
-			"node-ip-address": "$MY_POD_IP"})
+			"node-ip-address": "$MY_POD_IP",
+		})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Annotations, map[string]string{"annotation-1": "val1"})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Labels, map[string]string{"label-1": "val1"})
 	assert.Equal(t, ray.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.Tolerations, toleration)
 
 	workerReplica := int32(3)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Replicas, &workerReplica)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MinReplicas, &workerReplica)
-	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MaxReplicas, &workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Replicas, workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MinReplicas, workerReplica)
+	assert.Equal(t, *ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].MaxReplicas, workerReplica)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].GroupName, workerGroupName)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Spec.ServiceAccountName, serviceAccount)
 	assert.Equal(t, ray.Spec.RayClusterSpec.WorkerGroupSpecs[0].RayStartParams, map[string]string{"disable-usage-stats": "true", "node-ip-address": "$MY_POD_IP"})