Replace CHECK() with CHECK_*() so that when a check is failing, more …

heron/common/src/cpp/basics/basics.cpp

@@ -154,7 +154,7 @@ static void InitLogging() {
  *      Instance id of the program calling initialize
  */
 static void InitHelper(const char* argv0, const char* instance, bool istest) {
-  CHECK(signal(SIGPIPE, SIG_IGN) != SIG_ERR);
+  CHECK_NE(signal(SIGPIPE, SIG_IGN), SIG_ERR);
 
   // create execution meta data object
   SetMetadata(argv0, instance, istest);

heron/common/src/cpp/metrics/metricsmgr-client.cpp

@@ -142,7 +142,7 @@ void MetricsMgrClient::SendMetrics(proto::system::MetricPublisherPublishMessage*
 }
 
 void MetricsMgrClient::InternalSendTMasterLocation() {
-  CHECK(tmaster_location_);
+  CHECK(tmaster_location_) << "TMaster location is not available yet";
   proto::system::TMasterLocationRefreshMessage* m =
       new proto::system::TMasterLocationRefreshMessage();
   m->mutable_tmaster()->CopyFrom(*tmaster_location_);
@@ -152,7 +152,7 @@ void MetricsMgrClient::InternalSendTMasterLocation() {
 }
 
 void MetricsMgrClient::InternalSendMetricsCacheLocation() {
-  CHECK(metricscache_location_);
+  CHECK(metricscache_location_) << "MetricsCache location is not available yet";
   proto::system::MetricsCacheLocationRefreshMessage* m =
       new proto::system::MetricsCacheLocationRefreshMessage();
   m->mutable_metricscache()->CopyFrom(*metricscache_location_);

heron/common/src/cpp/network/baseconnection.cpp

@@ -56,9 +56,10 @@ BaseConnection::BaseConnection(ConnectionEndPoint* endpoint, ConnectionOptions*
 }
 
 BaseConnection::~BaseConnection() {
-  CHECK(mState == INIT || mState == DISCONNECTED);
-  bufferevent_free(buffer_);
+  CHECK(mState == INIT || mState == DISCONNECTED)
+      << "Deleting connection object while it is still connected";
   disableRateLimit();  // To free the config object
+  bufferevent_free(buffer_);
 }
 
 sp_int32 BaseConnection::start() {

heron/common/src/cpp/network/client.cpp

@@ -92,7 +92,8 @@ void Client::Init() { message_rid_gen_ = new REQID_Generator(); }
 
 void Client::InternalSendRequest(google::protobuf::Message* _request, void* _ctx, sp_int64 _msecs) {
   auto iter = requestResponseMap_.find(_request->GetTypeName());
-  CHECK(iter != requestResponseMap_.end());
+  CHECK(iter != requestResponseMap_.end())
+      << "Reponse type " << _request->GetTypeName() << " is not found";
   const sp_string& _expected_response_type = iter->second;
   if (state_ != CONNECTED) {
     delete _request;

heron/common/src/cpp/network/client.h

@@ -245,7 +245,7 @@ class Client : public BaseClient {
       __global_protobuf_pool_release__(m);
       return;
     }
-    CHECK(m->IsInitialized());
+    CHECK(m->IsInitialized()) << "Protobuf not initialized";
 
     std::function<void()> cb = std::bind(method, _t, m);
 

heron/common/src/cpp/network/httpserver.cpp

@@ -87,7 +87,7 @@ void HTTPServer::HandleHTTPRequest(struct evhttp_request* _request) {
 
 void HTTPServer::SendReply(IncomingHTTPRequest* _request, sp_int32 _code,
                            OutgoingHTTPResponse* _response) {
-  CHECK(_request->underlying_request() == _response->underlying_response());
+  CHECK_EQ(_request->underlying_request(), _response->underlying_response());
   evhttp_send_reply(_request->underlying_request(), _code, "", NULL);
   delete _response;
 }

heron/common/src/cpp/network/server.h

@@ -224,7 +224,7 @@ class Server : public BaseServer {
       CloseConnection(_conn);
       return;
     }
-    CHECK(m->IsInitialized());
+    CHECK(m->IsInitialized()) << "Protobuf not initialized";
 
     std::function<void()> cb = std::bind(method, _t, rid, _conn, m);
 

heron/common/src/cpp/zookeeper/zkclient.cpp

@@ -131,7 +131,7 @@ ZKClient::ZKClient(const std::string& hostportlist, EventLoop* eventLoop,
     : eventLoop_(eventLoop),
       hostportlist_(hostportlist),
       client_global_watcher_cb_(std::move(global_watcher_cb)) {
-  CHECK(client_global_watcher_cb_);
+  CHECK(client_global_watcher_cb_) << "Client global watcher is not provided";
   Init();
 }
 
@@ -341,7 +341,7 @@ void ZKClient::ZkWatcherCb(VCallback<> cb) {
 }
 
 void ZKClient::SendWatchEvent(const ZkWatchEvent& event) {
-  CHECK(client_global_watcher_cb_);
+  CHECK(client_global_watcher_cb_) << "Client global watcher is not provided";
   piper_->ExecuteInEventLoop(std::bind(&RunWatchEventCb, client_global_watcher_cb_, event));
 }
 

heron/instance/src/cpp/boltimpl/bolt-instance.cpp

@@ -70,7 +70,7 @@ BoltInstance::~BoltInstance() {
 }
 
 void BoltInstance::Start() {
-  CHECK(!active_);
+  CHECK(!active_) << "Bolt instance has been started already";
   LOG(INFO) << "Starting bolt " << taskContext_->getThisComponentName();
   bolt_->open(taskContext_->getConfig(), taskContext_, collector_);
   if (taskContext_->getConfig()->hasConfig(api::config::Config::TOPOLOGY_TICK_TUPLE_FREQ_SECS)) {
@@ -86,13 +86,13 @@ void BoltInstance::Start() {
 
 void BoltInstance::Activate() {
   LOG(INFO) << "Not doing anything in Bolt Activate";
-  CHECK(!active_);
+  CHECK(!active_) << "Bolt instance has been started already";
   active_ = true;
 }
 
 void BoltInstance::Deactivate() {
   LOG(INFO) << "Not doing anything in Bolt Dacctivate";
-  CHECK(active_);
+  CHECK(!active_) << "Bolt instance is not active";
   active_ = false;
 }
 

heron/instance/src/cpp/slave/imetrics-registrar-impl.cpp

@@ -76,7 +76,8 @@ void IMetricsRegistrarImpl::sendMetrics(int timeBucketSizeInSecs) {
       datum->set_name(metricName);
       datum->set_value(metricValue);
     } else {
-      CHECK(multiMetrics_.find(metricName) != multiMetrics_.end());
+      CHECK(multiMetrics_.find(metricName) != multiMetrics_.end())
+          << "Metric " << metricName << " is not found";
       std::map<std::string, std::string> mmap;
       multiMetrics_[metricName]->getValueAndReset(mmap);
       for (auto& kv : mmap) {

heron/instance/src/cpp/spoutimpl/spout-instance.cpp

@@ -73,7 +73,7 @@ SpoutInstance::~SpoutInstance() {
 }
 
 void SpoutInstance::Start() {
-  CHECK(!active_);
+  CHECK(!active_) << "Spout instance has been started already";
   LOG(INFO) << "Starting spout " << taskContext_->getThisComponentName()
             << " with ackingEnabled?: " << ackingEnabled_
             << " with enableMessageTimeouts?: " << enableMessageTimeouts_;
@@ -94,7 +94,7 @@ void SpoutInstance::Start() {
 
 void SpoutInstance::Activate() {
   LOG(INFO) << "Came in Activate of the spout";
-  CHECK(!active_);
+  CHECK(!active_) << "Spout instance has been started already";
   if (spout_) {
     spout_->activate();
   }
@@ -103,7 +103,7 @@ void SpoutInstance::Activate() {
 
 void SpoutInstance::Deactivate() {
   LOG(INFO) << "Came in Deactivate of the spout";
-  CHECK(active_);
+  CHECK(active_) << "Spout instance is not active";
   if (spout_) {
     spout_->deactivate();
   }

heron/statemgrs/src/cpp/statemgr/heron-localfilestatemgr.cpp

@@ -65,7 +65,7 @@ void HeronLocalFileStateMgr::InitTree() {
 
 void HeronLocalFileStateMgr::SetTMasterLocationWatch(const std::string& topology_name,
                                                      VCallback<> watcher) {
-  CHECK(watcher);
+  CHECK(watcher) << "Watcher callback is missing in SetTMasterLocationWatch()";
   // We kind of cheat here. We check periodically
   time_t tmaster_last_change = FileUtils::getModifiedTime(GetTMasterLocationPath(topology_name));
 
@@ -78,7 +78,7 @@ void HeronLocalFileStateMgr::SetTMasterLocationWatch(const std::string& topology
 
 void HeronLocalFileStateMgr::SetMetricsCacheLocationWatch(const std::string& topology_name,
                                                      VCallback<> watcher) {
-  CHECK(watcher);
+  CHECK(watcher) << "Watcher callback is missing in SetMetricsCacheLocationWatch()";
   // We kind of cheat here. We check periodically
   time_t tmaster_last_change = FileUtils::getModifiedTime(
                                GetMetricsCacheLocationPath(topology_name));
@@ -92,7 +92,7 @@ void HeronLocalFileStateMgr::SetMetricsCacheLocationWatch(const std::string& top
 
 void HeronLocalFileStateMgr::SetPackingPlanWatch(const std::string& topology_name,
                                                  VCallback<> watcher) {
-  CHECK(watcher);
+  CHECK(watcher) << "Watcher callback is missing in SetPackingPlanWatch";
   // We kind of cheat here. We check periodically
   time_t packingplan_last_change = FileUtils::getModifiedTime(GetPackingPlanPath(topology_name));
 

heron/statemgrs/src/cpp/statemgr/heron-zkstatemgr.cpp

@@ -74,31 +74,31 @@ HeronZKStateMgr::~HeronZKStateMgr() {
 
 void HeronZKStateMgr::InitTree() {
   // Needs to be implemented
-  CHECK(false);
+  CHECK(false) << "HeronZKStateMgr::InitTree() is not implemented yet";
 }
 
 void HeronZKStateMgr::SetTMasterLocationWatch(const std::string& topology_name,
                                               VCallback<> watcher) {
-  CHECK(watcher);
-  CHECK(!topology_name.empty());
+  CHECK(watcher) << "Watcher callback is missing in SetTMasterLocationWatch";
+  CHECK(!topology_name.empty()) << "Topology name is missing";
 
   tmaster_location_watcher_info_ = new TMasterLocationWatchInfo(std::move(watcher), topology_name);
   SetTMasterLocationWatchInternal();
 }
 
 void HeronZKStateMgr::SetMetricsCacheLocationWatch(const std::string& topology_name,
-                                              VCallback<> watcher) {
-  CHECK(watcher);
-  CHECK(!topology_name.empty());
+                                                   VCallback<> watcher) {
+  CHECK(watcher) << "Watcher callback is missing in SetMetricsCacheLocationWatch";
+  CHECK(!topology_name.empty()) << "Topology name is missing";
 
   metricscache_location_watcher_info_ = new TMasterLocationWatchInfo(
                               std::move(watcher), topology_name);
   SetMetricsCacheLocationWatchInternal();
 }
 
 void HeronZKStateMgr::SetPackingPlanWatch(const std::string& topology_name, VCallback<> watcher) {
-  CHECK(watcher);
-  CHECK(!topology_name.empty());
+  CHECK(watcher) << "Watcher callback is missing in SetPackingPlanWatch";
+  CHECK(!topology_name.empty()) << "Topology name is missing";
 
   packing_plan_watcher_info_ = new TMasterLocationWatchInfo(std::move(watcher), topology_name);
   SetPackingPlanWatchInternal();

heron/stmgr/src/cpp/grouping/fields-grouping.cpp

@@ -50,7 +50,7 @@ void FieldsGrouping::GetListToSend(const proto::system::HeronDataTuple& _tuple,
   size_t prime_num = 633910111UL;
   for (auto iter = fields_grouping_indices_.begin();
        iter != fields_grouping_indices_.end(); ++iter) {
-    CHECK(_tuple.values_size() > *iter);
+    CHECK_GT(_tuple.values_size(), *iter);
     size_t h = str_hash_fn(_tuple.values(*iter));
     task_index += (h % prime_num);
   }

heron/stmgr/src/cpp/manager/instance-server.cpp

@@ -222,7 +222,8 @@ void InstanceServer::HandleConnectionClose(Connection* _conn, NetworkErrorCode)
   auto iiter = active_instances_.find(_conn);
   if (iiter != active_instances_.end()) {
     sp_int32 task_id = iiter->second;
-    CHECK(instance_info_.find(task_id) != instance_info_.end());
+    CHECK(instance_info_.find(task_id) != instance_info_.end())
+        << "Task " << task_id << " is not found";
     sp_string instance_id = instance_info_[task_id]->instance_->instance_id();
     LOG(INFO) << "Instance " << instance_id << " closed connection";
 
@@ -524,7 +525,8 @@ void InstanceServer::StopBackPressureConnectionCb(Connection* _connection) {
   }
 
   CHECK(remote_ends_who_caused_back_pressure_.find(instance_name) !=
-        remote_ends_who_caused_back_pressure_.end());
+        remote_ends_who_caused_back_pressure_.end())
+      << "Instance " << instance_name << " is not found in the backpressure list";
   remote_ends_who_caused_back_pressure_.erase(instance_name);
 
   // Indicate which instance component stopped back pressure

heron/stmgr/src/cpp/manager/stateful-restorer.cpp

@@ -247,7 +247,8 @@ void StatefulRestorer::HandleAllInstancesConnected() {
 void StatefulRestorer::HandleDeadInstanceConnection(sp_int32 _task_id) {
   if (in_progress_) {
     instance_connections_pending_ = true;
-    CHECK(local_taskids_.find(_task_id) != local_taskids_.end());
+    CHECK(local_taskids_.find(_task_id) != local_taskids_.end())
+        << "Task " << _task_id << " is not found in local task list";
     restore_pending_.insert(_task_id);
     get_ckpt_pending_.insert(_task_id);
   }

heron/stmgr/src/cpp/manager/stmgr-clientmgr.cpp

@@ -131,7 +131,7 @@ StMgrClient* StMgrClientMgr::CreateClient(const sp_string& _other_stmgr_id,
 bool StMgrClientMgr::SendTupleStreamMessage(sp_int32 _task_id, const sp_string& _stmgr_id,
                                             const proto::system::HeronTupleSet2& _msg) {
   auto iter = clients_.find(_stmgr_id);
-  CHECK(iter != clients_.end());
+  CHECK(iter != clients_.end()) << "Stmgr " << _stmgr_id << " is not found in client list";
 
   // Acquire the message
   proto::stmgr::TupleStreamMessage* out = nullptr;
@@ -151,7 +151,7 @@ bool StMgrClientMgr::SendTupleStreamMessage(sp_int32 _task_id, const sp_string&
 void StMgrClientMgr::SendDownstreamStatefulCheckpoint(const sp_string& _stmgr_id,
                            proto::ckptmgr::DownstreamStatefulCheckpoint* _message) {
   auto iter = clients_.find(_stmgr_id);
-  CHECK(iter != clients_.end());
+  CHECK(iter != clients_.end()) << "Stmgr " << _stmgr_id << " is not found in client list";
   iter->second->SendDownstreamStatefulCheckpoint(_message);
 }
 

heron/stmgr/src/cpp/manager/stmgr-server.cpp

@@ -148,7 +148,8 @@ void StMgrServer::StartBackPressureClientCb(const sp_string& _other_stmgr_id) {
 
 void StMgrServer::StopBackPressureClientCb(const sp_string& _other_stmgr_id) {
   CHECK(remote_ends_who_caused_back_pressure_.find(_other_stmgr_id) !=
-        remote_ends_who_caused_back_pressure_.end());
+        remote_ends_who_caused_back_pressure_.end())
+      << "Stmgr " << _other_stmgr_id << " is not found in backpressure list";
   remote_ends_who_caused_back_pressure_.erase(_other_stmgr_id);
 
   if (!stmgr_->DidAnnounceBackPressure()) {
@@ -175,7 +176,8 @@ void StMgrServer::HandleStartBackPressureMessage(Connection* _conn,
     return;
   }
   auto iter = rstmgrs_.find(_conn);
-  CHECK(iter != rstmgrs_.end());
+  CHECK(iter != rstmgrs_.end())
+      << "Connection " << _conn << " is not found in remote stmgr list when starting backpressure";
   sp_string stmgr_id = iter->second;
   stmgrs_who_announced_back_pressure_.insert(stmgr_id);
 
@@ -196,7 +198,8 @@ void StMgrServer::HandleStopBackPressureMessage(Connection* _conn,
     return;
   }
   auto iter = rstmgrs_.find(_conn);
-  CHECK(iter != rstmgrs_.end());
+  CHECK(iter != rstmgrs_.end())
+      << "Connection " << _conn << " is not found in remote stmgr list when stopping backpressure";
   sp_string stmgr_id = iter->second;
   // Did we receive a start back pressure message from this stmgr to
   // begin with? We could have been dead at the time of the announcement

heron/stmgr/src/cpp/manager/stmgr.cpp

@@ -282,7 +282,7 @@ void StMgr::FetchMetricsCacheLocation() {
 }
 
 void StMgr::StartStmgrServer() {
-  CHECK(!stmgr_server_);
+  CHECK(!stmgr_server_) << "Stmgr server is already running";
   LOG(INFO) << "Creating StmgrServer";
   NetworkOptions sops;
   sops.set_host(IpUtils::getHostName());
@@ -302,7 +302,7 @@ void StMgr::StartStmgrServer() {
 }
 
 void StMgr::StartInstanceServer() {
-  CHECK(!instance_server_);
+  CHECK(!instance_server_) << "InstanceServer server is already running";
   LOG(INFO) << "Creating InstanceServer";
   NetworkOptions sops;
   sops.set_host(IpUtils::getHostName());
@@ -348,7 +348,7 @@ void StMgr::CreateCheckpointMgrClient() {
 }
 
 void StMgr::CreateTMasterClient(proto::tmaster::TMasterLocation* tmasterLocation) {
-  CHECK(!tmaster_client_);
+  CHECK(!tmaster_client_) << "TMaster client has already existed";
   LOG(INFO) << "Creating Tmaster Client at " << tmasterLocation->host() << ":"
             << tmasterLocation->master_port();
   NetworkOptions master_options;
@@ -384,7 +384,7 @@ void StMgr::CreateTMasterClient(proto::tmaster::TMasterLocation* tmasterLocation
 }
 
 void StMgr::CreateTupleCache() {
-  CHECK(!tuple_cache_);
+  CHECK(!tuple_cache_) << "Tuple cache has already existed";
   LOG(INFO) << "Creating tuple cache ";
   sp_uint32 drain_threshold_bytes_ =
       config::HeronInternalsConfigReader::Instance()->GetHeronStreammgrCacheDrainSizeMb() * 1_MB;
@@ -999,7 +999,7 @@ void StMgr::InitiateStatefulCheckpoint(sp_string _checkpoint_id) {
 void StMgr::HandleStoreInstanceStateCheckpoint(
             const proto::ckptmgr::InstanceStateCheckpoint& _message,
             const proto::system::Instance& _instance) {
-  CHECK(stateful_restorer_);
+  CHECK(stateful_restorer_) << "Stateful restorer doesn't exist when storing checkpoint";
   int32_t task_id = _instance.info().task_id();
   LOG(INFO) << "Got a checkpoint state message from " << task_id
             << " for checkpoint " << _message.checkpoint_id();
@@ -1072,7 +1072,7 @@ void StMgr::HandleDownStreamStatefulCheckpoint(
 void StMgr::RestoreTopologyState(sp_string _checkpoint_id, sp_int64 _restore_txid) {
   LOG(INFO) << "Got a Restore Topology State message from Tmaster for checkpoint "
             << _checkpoint_id << " and txid " << _restore_txid;
-  CHECK(stateful_restorer_);
+  CHECK(stateful_restorer_) << "Stateful restorer doesn't exist when restoring state";
 
   // Start the restore process
   std::unordered_set<sp_int32> local_taskids;
@@ -1085,7 +1085,7 @@ void StMgr::RestoreTopologyState(sp_string _checkpoint_id, sp_int64 _restore_txi
 void StMgr::StartStatefulProcessing(sp_string _checkpoint_id) {
   LOG(INFO) << "Received StartProcessing message from tmaster for "
             << _checkpoint_id;
-  CHECK(stateful_restorer_);
+  CHECK(stateful_restorer_) << "Stateful restorer doesn't exist when starting process";
   if (stateful_restorer_->InProgress()) {
     LOG(FATAL) << "StartProcessing received from Tmaster for "
                << _checkpoint_id << " when we are still in Restore";
@@ -1099,7 +1099,7 @@ void StMgr::HandleRestoreInstanceStateResponse(sp_int32 _task_id,
   // If we are stateful topology, we might want to see how the restore went
   // and if it was successful and all other local instances have recovered
   // send back a success response to tmaster saying that we have recovered
-  CHECK(stateful_restorer_);
+  CHECK(stateful_restorer_) << "Stateful restorer doesn't exist when querying restore state";
   stateful_restorer_->HandleInstanceRestoredState(_task_id, _status.status(), _checkpoint_id);
 }
 

heron/stmgr/src/cpp/util/tuple-cache.cpp

@@ -125,8 +125,8 @@ TupleCache::TupleList::TupleList() {
 }
 
 TupleCache::TupleList::~TupleList() {
-  CHECK(tuples_.empty());
-  CHECK(!current_);
+  CHECK(tuples_.empty()) << "Buffer is not empty when destructing TupleList";
+  CHECK(!current_) << "Current tuple set is not empty when destructing TupleList";
 }
 
 void TupleCache::TupleList::clear() {