Revert "Revert "[Core] RetryObjectInPlasmaErrors tries to fetch all o…

…bjects, not just ready ones." (ray-project#31445)" (ray-project#34805)

This reverts commit a48e8d7.

src/ray/core_worker/core_worker.cc

@@ -1452,20 +1452,11 @@ void RetryObjectInPlasmaErrors(std::shared_ptr<CoreWorkerMemoryStore> &memory_st
  for (auto iter = memory_object_ids.begin(); iter != memory_object_ids.end();) {
  auto current = iter++;
  const auto &mem_id = *current;
- auto ready_iter = ready.find(mem_id);
- if (ready_iter != ready.end()) {
- std::vector<std::shared_ptr<RayObject>> found;
- RAY_CHECK_OK(memory_store->Get({mem_id},
- /*num_objects=*/1,
- /*timeout=*/0,
- worker_context,
- /*remote_after_get=*/false,
- &found));
- if (found.size() == 1 && found[0]->IsInPlasmaError()) {
- plasma_object_ids.insert(mem_id);
- ready.erase(ready_iter);
- memory_object_ids.erase(current);
- }
+ auto found = memory_store->GetIfExists(mem_id);
+ if (found != nullptr && found->IsInPlasmaError()) {
+ plasma_object_ids.insert(mem_id);
+ ready.erase(mem_id);
+ memory_object_ids.erase(current);
  }
  }
 }