[7.x][ML] Fixing memory_status output on fatal error (#1438)

When the job is failing due to the memory limit exceeding, the job fails to emit the final memory_status message. This PR fixes this.

There was a type in the "hard_limit" tag which I fixed. Also I tweaked the parameters of the unit test CDataFrameAnalyzerTrainingTest.testMemoryLimitHandling to reduce runtime.

Additionally I took the new messaging from #1428 and added it here to avoid merge conflicts.

Author: valeriy42

lib/api/CDataFrameAnalysisInstrumentation.cc

@@ -40,7 +40,7 @@ const std::string HYPERPARAMETERS_TAG{"hyperparameters"};
 const std::string MEMORY_REESTIMATE_TAG{"memory_reestimate_bytes"};
 const std::string ITERATION_TAG{"iteration"};
 const std::string JOB_ID_TAG{"job_id"};
-const std::string MEMORY_STATUS_HARD_LIMIT_TAG{"hard-limit"};
+const std::string MEMORY_STATUS_HARD_LIMIT_TAG{"hard_limit"};
 const std::string MEMORY_STATUS_OK_TAG{"ok"};
 const std::string MEMORY_STATUS_TAG{"status"};
 const std::string MEMORY_TYPE_TAG{"analytics_memory_usage"};
@@ -190,10 +190,12 @@ void CDataFrameAnalysisInstrumentation::monitor(CDataFrameAnalysisInstrumentatio
             instrumentation.memoryReestimate(static_cast<std::int64_t>(memoryReestimateBytes));
             instrumentation.memoryStatus(E_HardLimit);
             instrumentation.flush();
-            HANDLE_FATAL(<< "Input error: required memory " << bytesToString(memory)
-                         << " exceeds the memory limit " << bytesToString(memoryLimit)
-                         << ". Please force-stop the analysis job, increase the limit to at least "
-                         << bytesToString(memoryReestimateBytes) << " and restart.")
+            writer.flush();
+            LOG_INFO(<< "Required memory " << memory << " exceeds the memory limit " << memoryLimit
+                     << ".  New estimated limit is " << memoryReestimateBytes << ".");
+            HANDLE_FATAL(<< "Input error: memory limit [" << bytesToString(memoryLimit)
+                         << "] has been exceeded. Please force stop the job, increase to new estimated limit ["
+                         << bytesToString(memoryReestimateBytes) << "] and restart.")
         }
 
         wait = std::min(2 * wait, 1024);

lib/api/unittest/CDataFrameAnalyzerOutlierTest.cc

@@ -530,7 +530,7 @@ BOOST_AUTO_TEST_CASE(testErrors) {
         BOOST_TEST_REQUIRE(errors.size() > 0);
         bool memoryLimitExceed{false};
         for (const auto& error : errors) {
-            if (error.find("Input error: required memory") != std::string::npos) {
+            if (error.find("Input error: memory limit") != std::string::npos) {
                 memoryLimitExceed = true;
                 break;
             }
@@ -549,7 +549,7 @@ BOOST_AUTO_TEST_CASE(testErrors) {
                 std::string status{result["analytics_memory_usage"]["status"].GetString()};
                 if (status == "ok") {
                     memoryStatusOk = true;
-                } else if (status == "hard-limit") {
+                } else if (status == "hard_limit") {
                     memoryStatusHardLimit = true;
                     if (result["analytics_memory_usage"].HasMember("memory_reestimate_bytes") &&
                         result["analytics_memory_usage"]["memory_reestimate_bytes"]

lib/api/unittest/CDataFrameAnalyzerTrainingTest.cc

@@ -378,23 +378,28 @@ BOOST_AUTO_TEST_CASE(testMemoryLimitHandling) {
     };
 
     TDoubleVec expectedPredictions;
+    std::size_t numberSamples{50};
 
     TStrVec fieldNames{"f1", "f2", "f3", "f4", "target", ".", "."};
     TStrVec fieldValues{"", "", "", "", "", "0", ""};
     api::CDataFrameAnalyzer analyzer{
-        test::CDataFrameAnalysisSpecificationFactory{}.memoryLimit(1000).predictionSpec(
-            test::CDataFrameAnalysisSpecificationFactory::regression(), "target"),
+        test::CDataFrameAnalysisSpecificationFactory{}
+            .rows(numberSamples)
+            .predictionMaximumNumberTrees(2)
+            .memoryLimit(10)
+            .predicitionNumberRoundsPerHyperparameter(1)
+            .predictionSpec(test::CDataFrameAnalysisSpecificationFactory::regression(), "target"),
         outputWriterFactory};
     test::CDataFrameAnalyzerTrainingFactory::addPredictionTestData(
         TLossFunctionType::E_MseRegression, fieldNames, fieldValues, analyzer,
-        expectedPredictions);
+        expectedPredictions, numberSamples);
 
     analyzer.handleRecord(fieldNames, {"", "", "", "", "", "", "$"});
 
     BOOST_TEST_REQUIRE(errors.size() > 0);
     bool memoryLimitExceed{false};
     for (const auto& error : errors) {
-        if (error.find("Input error: required memory") != std::string::npos) {
+        if (error.find("Input error: memory limit") != std::string::npos) {
             memoryLimitExceed = true;
             break;
         }
@@ -413,7 +418,7 @@ BOOST_AUTO_TEST_CASE(testMemoryLimitHandling) {
             std::string status{result["analytics_memory_usage"]["status"].GetString()};
             if (status == "ok") {
                 memoryStatusOk = true;
-            } else if (status == "hard-limit") {
+            } else if (status == "hard_limit") {
                 memoryStatusHardLimit = true;
                 if (result["analytics_memory_usage"].HasMember("memory_reestimate_bytes") &&
                     result["analytics_memory_usage"]["memory_reestimate_bytes"].GetInt() > 0) {

lib/core/CRapidJsonConcurrentLineWriter.cc

@@ -15,6 +15,7 @@ CRapidJsonConcurrentLineWriter::CRapidJsonConcurrentLineWriter(CJsonOutputStream
 }
 
 CRapidJsonConcurrentLineWriter::~CRapidJsonConcurrentLineWriter() {
+    this->flush();
     m_OutputStreamWrapper.releaseBuffer(*this, m_StringBuffer);
 }