Add data quality field in GaaSJobObservabilityEventSchema (#4114)

* Add data quality field in GaaSJobObservabilityEventSchema

---------

Co-authored-by: Vaibhav Singhal <vaibsing@vaibsing-mn7618.linkedin.biz>

Author: vsinghal85

gobblin-core/src/main/java/org/apache/gobblin/qualitychecker/task/TaskLevelPolicyChecker.java

@@ -29,6 +29,16 @@
  * in a PolicyCheckResults object
  */
 public class TaskLevelPolicyChecker {
+  /**
+   * An enumeration for possible statuses for Data quality checks,
+   * its values will be PASSED, FAILED, in case if data quality check
+   * evaluation is not performed for Job, it will be NOT_EVALUATED
+   */
+  public enum DataQualityStatus {
+    PASSED,
+    FAILED,
+    NOT_EVALUATED
+  }
   private final List<TaskLevelPolicy> list;
   private static final Logger LOG = LoggerFactory.getLogger(TaskLevelPolicyChecker.class);
 

gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc

@@ -258,6 +258,12 @@
                 "name": "successfullyCommitted",
                 "type": "boolean",
                 "doc": "Whether the dataset was successfully committed by Gobblin and fully successful, versus when users configure partial failures or non-atomic writes"
+              },
+              {
+                "name": "dataQualityStatus",
+                "type": ["null","string"],
+                "doc": "Whether the dataset passed the overall data quality check",
+                "default": null
               }
             ]
           }

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java

@@ -25,6 +25,7 @@
 import lombok.Setter;
 
 import org.apache.gobblin.metrics.DatasetMetric;
+import org.apache.gobblin.qualitychecker.task.TaskLevelPolicyChecker;
 
 
 /**
@@ -45,6 +46,7 @@ public class DatasetTaskSummary {
    * Convert a {@link DatasetTaskSummary} to a {@link DatasetMetric}.
    */
   public static DatasetMetric toDatasetMetric(DatasetTaskSummary datasetTaskSummary) {
-    return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted());
+    return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(),
+        TaskLevelPolicyChecker.DataQualityStatus.NOT_EVALUATED.name());
   }
 }