fix java coreworker crash (#9674)

src/ray/core_worker/actor_manager.cc

@@ -89,9 +89,11 @@ bool ActorManager::AddActorHandle(std::unique_ptr<ActorHandle> actor_handle,
     auto actor_notification_callback =
         std::bind(&ActorManager::HandleActorStateNotification, this,
                   std::placeholders::_1, std::placeholders::_2);
-
-    RAY_CHECK_OK(gcs_client_->Actors().AsyncSubscribe(
-        actor_id, actor_notification_callback, nullptr));
+    {
+      absl::MutexLock lock(&gcs_client_mutex_);
+      RAY_CHECK_OK(gcs_client_->Actors().AsyncSubscribe(
+          actor_id, actor_notification_callback, nullptr));
+    }
 
     if (!RayConfig::instance().gcs_actor_service_enabled()) {
       RAY_CHECK(reference_counter_->SetDeleteCallback(

src/ray/core_worker/actor_manager.h

@@ -132,8 +132,11 @@ class ActorManager {
   void HandleActorStateNotification(const ActorID &actor_id,
                                     const gcs::ActorTableData &actor_data);
 
-  /// GCS client
-  std::shared_ptr<gcs::GcsClient> gcs_client_;
+  /// Mutex to protect the gcs_client_ field.
+  /// NOTE: Now gcs client is not thread safe, so we add lock protection.
+  mutable absl::Mutex gcs_client_mutex_;
+  /// GCS client.
+  std::shared_ptr<gcs::GcsClient> gcs_client_ GUARDED_BY(gcs_client_mutex_);
 
   /// Interface to submit tasks directly to other actors.
   std::shared_ptr<CoreWorkerDirectActorTaskSubmitterInterface> direct_actor_submitter_;