phet commented on code in PR #3912:
URL: https://github.com/apache/gobblin/pull/3912#discussion_r1562918956
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/CommitActivityImpl.java:
##########
@@ -181,6 +195,40 @@ public static Map<String, JobState.DatasetState>
createDatasetStatesByUrns(Colle
return datasetStatesByUrns;
}
+ public List<DatasetTaskSummary> generateDatasetTaskSummaries(Map<String,
JobState.DatasetState> datasetStatesByUrns, JobContext jobContext,
EventSubmitter eventSubmitter) {
+ List<DatasetTaskSummary> datasetTaskSummaries = new ArrayList<>();
+ // Only process successful datasets unless configuration to process failed
datasets is set
+ boolean processFailedTasks =
+
PropertiesUtils.getPropAsBoolean(jobContext.getJobState().getProperties(),
ConfigurationKeys.WRITER_COUNT_METRICS_FROM_FAILED_TASKS,
+ "false");
+ for (JobState.DatasetState datasetState : datasetStatesByUrns.values()) {
+ if (datasetState.getState() == JobState.RunningState.COMMITTED ||
(datasetState.getState() == JobState.RunningState.FAILED
+ && jobContext.getJobCommitPolicy() ==
JobCommitPolicy.COMMIT_SUCCESSFUL_TASKS)) {
+ long totalBytesWritten = 0;
+ long totalRecordsWritten = 0;
+ for (TaskState taskState : datasetState.getTaskStates()) {
+ // Certain writers may omit these metrics e.g.
CompactionLauncherWriter
+ if ((taskState.getWorkingState() ==
WorkUnitState.WorkingState.COMMITTED || processFailedTasks)) {
+ totalBytesWritten +=
taskState.getPropAsLong(ConfigurationKeys.WRITER_BYTES_WRITTEN, 0);
+ totalRecordsWritten +=
taskState.getPropAsLong(ConfigurationKeys.WRITER_RECORDS_WRITTEN, 0);
+ }
+ }
+ log.info(String.format("DatasetMetrics for '%s' - (records: %d; bytes:
%d)", datasetState.getDatasetUrn(),
+ totalRecordsWritten, totalBytesWritten));
+ datasetTaskSummaries.add(
+ new DatasetTaskSummary(datasetState.getDatasetUrn(),
totalRecordsWritten, totalBytesWritten,
+ datasetState.getState() == JobState.RunningState.COMMITTED));
+ } else if (datasetState.getState() == JobState.RunningState.FAILED) {
+ // Check if config is turned on for submitting writer metrics on
failure due to non-atomic write semantics
+ if (jobContext.getJobCommitPolicy() ==
JobCommitPolicy.COMMIT_ON_FULL_SUCCESS) {
Review Comment:
for clarity (and to parallel the `&&` in the `if` alternative above),
suggest to combine the nested `if` via `&&`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
