Log fatal error if plasma manager or local scheduler heartbeats take too long.
diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt
index 303ecf4ab4eaf..27cb1a969c8e1 100644
--- a/src/common/CMakeLists.txt
+++ b/src/common/CMakeLists.txt
@@ -79,7 +79,7 @@ function(define_test test_name library)
   add_executable(${test_name} test/${test_name}.cc ${ARGN})
   add_dependencies(${test_name} hiredis flatbuffers_ep)
   target_link_libraries(${test_name} common ${FLATBUFFERS_STATIC_LIB} ${library})
-  target_compile_options(${test_name} PUBLIC "-DPLASMA_TEST -DLOCAL_SCHEDULER_TEST -DCOMMON_TEST -DRAY_COMMON_LOG_LEVEL=4 -DRAY_TIMEOUT=50")
+  target_compile_options(${test_name} PUBLIC "-DPLASMA_TEST -DLOCAL_SCHEDULER_TEST -DCOMMON_TEST -DRAY_COMMON_LOG_LEVEL=4")
 endfunction()
 
 define_test(common_tests "")
diff --git a/src/common/common.cc b/src/common/common.cc
index 91b36797cad9b..e032ae3ecf8ec 100644
--- a/src/common/common.cc
+++ b/src/common/common.cc
@@ -1,5 +1,6 @@
 #include "common.h"
 
+#include <chrono>
 #include <stdio.h>
 #include <unistd.h>
 #include <sys/types.h>
@@ -65,3 +66,10 @@ char *ObjectID_to_string(ObjectID obj_id, char *id_string, int id_length) {
 
   return id_string;
 }
+
+int64_t current_time_ms() {
+  std::chrono::milliseconds ms_since_epoch =
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          std::chrono::system_clock::now().time_since_epoch());
+  return ms_since_epoch.count();
+}
diff --git a/src/common/common.h b/src/common/common.h
index 9560123d7ab84..6c09913acbbd9 100644
--- a/src/common/common.h
+++ b/src/common/common.h
@@ -232,4 +232,11 @@ bool DBClientID_equal(DBClientID first_id, DBClientID second_id);
 
 extern const unsigned char NIL_DIGEST[DIGEST_SIZE];
 
+/**
+ * Return the current time in milliseconds since the Unix epoch.
+ *
+ * @return The number of milliseconds since the Unix epoch.
+ */
+int64_t current_time_ms();
+
 #endif
diff --git a/src/local_scheduler/local_scheduler.cc b/src/local_scheduler/local_scheduler.cc
index f6500d3e0d408..1454f5e1575aa 100644
--- a/src/local_scheduler/local_scheduler.cc
+++ b/src/local_scheduler/local_scheduler.cc
@@ -405,6 +405,9 @@ LocalSchedulerState *LocalSchedulerState_init(
     start_worker(state, NIL_ACTOR_ID);
   }
 
+  /* Initialize the time at which the previous heartbeat was sent. */
+  state->previous_heartbeat_time = current_time_ms();
+
   return state;
 }
 
@@ -1122,6 +1125,17 @@ void handle_actor_creation_callback(ActorID actor_id,
 int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   LocalSchedulerState *state = (LocalSchedulerState *) context;
   SchedulingAlgorithmState *algorithm_state = state->algorithm_state;
+
+  /* Check that the last heartbeat was not sent too long ago. */
+  int64_t current_time = current_time_ms();
+  CHECK(current_time >= state->previous_heartbeat_time);
+  if (current_time - state->previous_heartbeat_time >
+      NUM_HEARTBEATS_TIMEOUT * HEARTBEAT_TIMEOUT_MILLISECONDS) {
+    LOG_FATAL("The last heartbeat was sent %" PRId64 " milliseconds ago.",
+              current_time - state->previous_heartbeat_time);
+  }
+  state->previous_heartbeat_time = current_time;
+
   LocalSchedulerInfo info;
   /* Ask the scheduling algorithm to fill out the scheduler info struct. */
   provide_scheduler_info(state, algorithm_state, &info);
diff --git a/src/local_scheduler/local_scheduler_shared.h b/src/local_scheduler/local_scheduler_shared.h
index 94627b7abeb3d..8b35c5ec4d91d 100644
--- a/src/local_scheduler/local_scheduler_shared.h
+++ b/src/local_scheduler/local_scheduler_shared.h
@@ -78,6 +78,9 @@ struct LocalSchedulerState {
    *  available_gpus.size() == dynamic_resources[ResourceIndex_GPU] should
    *  always be true. */
   std::vector<int> available_gpus;
+  /** The time (in milliseconds since the Unix epoch) when the most recent
+   *  heartbeat was sent. */
+  int64_t previous_heartbeat_time;
 };
 
 /** Contains all information associated with a local scheduler client. */
diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc
index e9308a29570b5..b54cb8fdc5028 100644
--- a/src/plasma/plasma_manager.cc
+++ b/src/plasma/plasma_manager.cc
@@ -236,6 +236,9 @@ struct PlasmaManagerState {
   ObjectWaitRequests *object_wait_requests_remote;
   /** Initialize an empty hash map for the cache of local available object. */
   AvailableObject *local_available_objects;
+  /** The time (in milliseconds since the Unix epoch) when the most recent
+   *  heartbeat was sent. */
+  int64_t previous_heartbeat_time;
 };
 
 PlasmaManagerState *g_manager_state = NULL;
@@ -553,6 +556,8 @@ PlasmaManagerState *PlasmaManagerState_init(const char *store_socket_name,
   /* Add the callback that processes the notification to the event loop. */
   event_loop_add_file(state->loop, plasma_fd, EVENT_LOOP_READ,
                       process_object_notification, state);
+  /* Initialize the time at which the previous heartbeat was sent. */
+  state->previous_heartbeat_time = current_time_ms();
   return state;
 }
 
@@ -1590,6 +1595,17 @@ void process_message(event_loop *loop,
 
 int heartbeat_handler(event_loop *loop, timer_id id, void *context) {
   PlasmaManagerState *state = (PlasmaManagerState *) context;
+
+  /* Check that the last heartbeat was not sent too long ago. */
+  int64_t current_time = current_time_ms();
+  CHECK(current_time >= state->previous_heartbeat_time);
+  if (current_time - state->previous_heartbeat_time >
+      NUM_HEARTBEATS_TIMEOUT * HEARTBEAT_TIMEOUT_MILLISECONDS) {
+    LOG_FATAL("The last heartbeat was sent %" PRId64 " milliseconds ago.",
+              current_time - state->previous_heartbeat_time);
+  }
+  state->previous_heartbeat_time = current_time;
+
   plasma_manager_send_heartbeat(state->db);
   return HEARTBEAT_TIMEOUT_MILLISECONDS;
 }
diff --git a/src/plasma/plasma_manager.h b/src/plasma/plasma_manager.h
index f28224570c57e..e9247a4f15a83 100644
--- a/src/plasma/plasma_manager.h
+++ b/src/plasma/plasma_manager.h
@@ -12,11 +12,9 @@
 #endif
 
 /* Timeouts are in milliseconds. */
-#ifndef RAY_TIMEOUT
 #define MANAGER_TIMEOUT 1000
-#else
-#define MANAGER_TIMEOUT RAY_TIMEOUT
-#endif
+
+#define NUM_HEARTBEATS_TIMEOUT 100
 
 /* The buffer size in bytes. Data will get transfered in multiples of this */
 #define BUFSIZE 4096
