Initialize client_id_ in ObjectManager constructor that takes user-defined ObjectDirectory (#3403)

src/ray/object_manager/object_directory.cc

@@ -192,6 +192,10 @@ ray::Status ObjectDirectory::LookupLocations(const ObjectID &object_id,
   return status;
 }
 
+ray::ClientID ObjectDirectory::GetLocalClientID() {
+  return gcs_client_->client_table().GetLocalClientId();
+}
+
 std::string ObjectDirectory::DebugString() const {
   std::stringstream result;
   result << "ObjectDirectory:";

src/ray/object_manager/object_directory.h

@@ -104,6 +104,11 @@ class ObjectDirectoryInterface {
   virtual ray::Status ReportObjectRemoved(const ObjectID &object_id,
                                           const ClientID &client_id) = 0;
 
+  /// Get local client id
+  ///
+  /// \return ClientID
+  virtual ray::ClientID GetLocalClientID() = 0;
+
   /// Returns debug string for class.
   ///
   /// \return string.
@@ -145,6 +150,8 @@ class ObjectDirectory : public ObjectDirectoryInterface {
   ray::Status ReportObjectRemoved(const ObjectID &object_id,
                                   const ClientID &client_id) override;
 
+  ray::ClientID GetLocalClientID() override;
+
   std::string DebugString() const override;
 
   /// ObjectDirectory should not be copied.

src/ray/object_manager/object_manager.cc

@@ -52,7 +52,7 @@ ObjectManager::ObjectManager(asio::io_service &main_service,
       gen_(std::chrono::high_resolution_clock::now().time_since_epoch().count()) {
   RAY_CHECK(config_.max_sends > 0);
   RAY_CHECK(config_.max_receives > 0);
-  // TODO(hme) Client ID is never set with this constructor.
+  client_id_ = object_directory_->GetLocalClientID();
   main_service_ = &main_service;
   store_notification_.SubscribeObjAdded(
       [this](const object_manager::protocol::ObjectInfoT &object_info) {

src/ray/raylet/reconstruction_policy_test.cc

@@ -39,6 +39,7 @@ class MockObjectDirectory : public ObjectDirectoryInterface {
   std::string DebugString() const { return ""; }
 
   MOCK_METHOD0(RegisterBackend, void(void));
+  MOCK_METHOD0(GetLocalClientID, ray::ClientID());
   MOCK_CONST_METHOD1(LookupRemoteConnectionInfo, void(RemoteConnectionInfo &));
   MOCK_CONST_METHOD0(LookupAllRemoteConnections, std::vector<RemoteConnectionInfo>());
   MOCK_METHOD3(SubscribeObjectLocations,