From 17da49d9387529010b8eb3f5fb2236d40495cbfe Mon Sep 17 00:00:00 2001 From: pratapaditya04 Date: Wed, 22 Jul 2026 10:42:04 +0530 Subject: [PATCH 1/3] [GOBBLIN-XXXX] Surface snapshot/partition copy metrics in observability Add per-dataset snapshotsCommitted and partitionsCommitted counts to both the GaaSJobObservabilityEvent DatasetMetric and the job-status JobStatistics REST surface, so file/blob movement jobs (e.g. DDM Iceberg snapshot replication) can report table snapshots and partitions committed. DatasetMetric (GaaSJobObservabilityEvent.avsc) gains snapshotsCommitted and partitionsCommitted (long, default -1 = unsupported/unknown). DatasetTaskSummary carries the two counts as non-@NonNull fields so the existing 5-arg RequiredArgsConstructor used by AbstractJobLauncher is preserved; they populate via JSON deserialization for events that supply them, and toDatasetMetric maps them through. JobStatistics (executionStatistics) gains optional bytesWritten, recordsWritten, filesCommitted, snapshotsCommitted and partitionsCommitted. JobStatusRetriever reads them from the flat ddm.* copy-metric keys the JobSummary event persists into the job state, and FlowExecutionResource sets each only when reported, so jobs that do not emit them are unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../main/avro/GaaSJobObservabilityEvent.avsc | 12 +++++++ .../apache/gobblin/service/JobStatistics.pdl | 31 +++++++++++++++++ ...bblin.service.flowexecutions.snapshot.json | 33 +++++++++++++++++++ ...gobblin.service.flowstatuses.snapshot.json | 25 ++++++++++++++ .../service/FlowExecutionResource.java | 30 +++++++++++++---- .../gobblin/runtime/DatasetTaskSummary.java | 8 ++++- .../gobblin/service/monitoring/JobStatus.java | 8 +++++ .../monitoring/JobStatusRetriever.java | 19 ++++++++++- 8 files changed, 158 insertions(+), 8 deletions(-) diff --git a/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc b/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc index 53671f84170..165560f9be9 100644 --- a/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc +++ b/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc @@ -264,6 +264,18 @@ "type": ["null","string"], "doc": "Whether the dataset passed the overall data quality check", "default": null + }, + { + "name": "snapshotsCommitted", + "type": ["null","string"], + "doc": "Comma-separated list of the ids of the table snapshots committed for the dataset in this run (e.g. Iceberg snapshot-replication commits), null if unsupported/unreported by the writer", + "default": null + }, + { + "name": "partitionsCommitted", + "type": ["null","string"], + "doc": "Comma-separated list of the partitions committed for the dataset in this run, null if the writer does not commit partitions or does not report them", + "default": null } ] } diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl index c0c072e3065..9644824387f 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl @@ -29,4 +29,35 @@ record JobStatistics { * estimate of time left until job completion */ estimatedSecondsToCompletion: long + + /** + * Number of bytes written/copied for the dataset in this job execution. + * Optional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported. + */ + bytesWritten: optional long + + /** + * Number of records (or files, for file/blob copies) written for the dataset in this job execution. + * Optional; -1 if unsupported by the writer. + */ + recordsWritten: optional long + + /** + * Number of files committed for the dataset in this job execution. + * Optional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise. + */ + filesCommitted: optional long + + /** + * Comma-separated list of the ids of the table snapshots committed for the dataset in this job + * execution (e.g. Iceberg snapshot-replication commits). + * Optional; absent when the writer does not commit snapshots or does not report them. + */ + snapshotsCommitted: optional string + + /** + * Comma-separated list of the partitions committed for the dataset in this job execution. + * Optional; absent when the writer does not commit partitions or does not report them. + */ + partitionsCommitted: optional string } \ No newline at end of file diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json index b2cdddc5e3a..57cdd9c7c3e 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json @@ -186,6 +186,31 @@ "name" : "estimatedSecondsToCompletion", "type" : "long", "doc" : "estimate of time left until job completion" + }, { + "name" : "bytesWritten", + "type" : "long", + "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.", + "optional" : true + }, { + "name" : "recordsWritten", + "type" : "long", + "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.", + "optional" : true + }, { + "name" : "filesCommitted", + "type" : "long", + "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.", + "optional" : true + }, { + "name" : "snapshotsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.", + "optional" : true + }, { + "name" : "partitionsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.", + "optional" : true } ] }, "doc" : "Statistics from the job execution. The values may be updated during the run of a job." @@ -258,6 +283,14 @@ } }, "doc" : "Status of jobs belonging to the flow" + }, { + "name" : "issues", + "type" : { + "type" : "array", + "items" : "Issue" + }, + "doc" : "Collection of flow-level issues from the orchestration layer.\nPopulated when errors occur before any job starts (e.g., compilation failures)\nor when an error affects the entire flow (e.g., flow SLA exceeded).", + "default" : [ ] } ] }, "org.apache.gobblin.service.FlowId", "org.apache.gobblin.service.FlowStatistics", "org.apache.gobblin.service.FlowStatusId", "org.apache.gobblin.service.Issue", "org.apache.gobblin.service.IssueSeverity", "org.apache.gobblin.service.JobId", "org.apache.gobblin.service.JobState", "org.apache.gobblin.service.JobStatistics", "org.apache.gobblin.service.JobStatus", "org.apache.gobblin.service.Timestamp" ], "schema" : { diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json index df4d02067a5..aba90d9dac1 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json @@ -188,6 +188,31 @@ "name" : "estimatedSecondsToCompletion", "type" : "long", "doc" : "estimate of time left until job completion" + }, { + "name" : "bytesWritten", + "type" : "long", + "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.", + "optional" : true + }, { + "name" : "recordsWritten", + "type" : "long", + "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.", + "optional" : true + }, { + "name" : "filesCommitted", + "type" : "long", + "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.", + "optional" : true + }, { + "name" : "snapshotsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.", + "optional" : true + }, { + "name" : "partitionsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.", + "optional" : true } ] }, "doc" : "Statistics from the job execution. The values may be updated during the run of a job." diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java index 0bcdf569430..23dcbd8fd51 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java @@ -184,17 +184,35 @@ public static FlowExecution convertFlowStatus(FlowStatus monitoringFlowStatus, Long timeLeft = estimateCopyTimeLeft(queriedJobStatus.getLastProgressEventTime(), queriedJobStatus.getStartTime(), queriedJobStatus.getProgressPercentage()); + JobStatistics jobStatistics = new JobStatistics() + .setExecutionStartTime(queriedJobStatus.getStartTime()) + .setExecutionEndTime(queriedJobStatus.getEndTime()) + .setProcessedCount(queriedJobStatus.getProcessedCount()) + .setJobProgress(queriedJobStatus.getProgressPercentage()) + .setEstimatedSecondsToCompletion(timeLeft); + // Optional per-dataset copy metrics (e.g. DDM file/blob replication); set only when reported. + if (queriedJobStatus.getBytesWritten() >= 0) { + jobStatistics.setBytesWritten(queriedJobStatus.getBytesWritten()); + } + if (queriedJobStatus.getRecordsWritten() >= 0) { + jobStatistics.setRecordsWritten(queriedJobStatus.getRecordsWritten()); + } + if (queriedJobStatus.getFilesCommitted() >= 0) { + jobStatistics.setFilesCommitted(queriedJobStatus.getFilesCommitted()); + } + if (queriedJobStatus.getSnapshotsCommitted() != null) { + jobStatistics.setSnapshotsCommitted(queriedJobStatus.getSnapshotsCommitted()); + } + if (queriedJobStatus.getPartitionsCommitted() != null) { + jobStatistics.setPartitionsCommitted(queriedJobStatus.getPartitionsCommitted()); + } + jobStatus.setFlowId(flowId) .setJobId(new JobId() .setJobName(queriedJobStatus.getJobName()) .setJobGroup(queriedJobStatus.getJobGroup())) .setJobTag(queriedJobStatus.getJobTag(), SetMode.IGNORE_NULL) - .setExecutionStatistics(new JobStatistics() - .setExecutionStartTime(queriedJobStatus.getStartTime()) - .setExecutionEndTime(queriedJobStatus.getEndTime()) - .setProcessedCount(queriedJobStatus.getProcessedCount()) - .setJobProgress(queriedJobStatus.getProgressPercentage()) - .setEstimatedSecondsToCompletion(timeLeft)) + .setExecutionStatistics(jobStatistics) .setExecutionStatus(ExecutionStatus.valueOf(queriedJobStatus.getEventName())) .setMessage(queriedJobStatus.getMessage()) .setJobState(new JobState() diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java index f8c1d2a1e08..88a493077e5 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java @@ -42,11 +42,17 @@ public class DatasetTaskSummary { @NonNull private long bytesWritten; @NonNull private boolean successfullyCommitted; @NonNull private String dataQualityStatus; + // NOTE: intentionally NOT @NonNull so the 5-arg @RequiredArgsConstructor stays intact for native + // Gobblin call sites (AbstractJobLauncher). Populated via reflection during JSON deserialization + // of events that carry them (e.g. DDM Iceberg snapshot replication). Both are comma-separated + // lists (a run can commit more than one snapshot/partition); null = unsupported/unreported. + private String snapshotsCommitted = null; + private String partitionsCommitted = null; /** * 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(), datasetTaskSummary.getDataQualityStatus()); + return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(), datasetTaskSummary.getDataQualityStatus(), datasetTaskSummary.getSnapshotsCommitted(), datasetTaskSummary.getPartitionsCommitted()); } } diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java index 8dd91139914..a52cf0e874b 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java @@ -60,4 +60,12 @@ public class JobStatus { private final Supplier> issues; private final int progressPercentage; private final long lastProgressEventTime; + // Optional per-dataset copy metrics surfaced on the job-status REST API (JobStatistics). + // Populated for writers that report them (e.g. DDM file/blob replication); -1 = unset/unsupported + // for the counts. The snapshot ids and partitions are comma-separated lists; null = unreported. + @Builder.Default private final long bytesWritten = -1L; + @Builder.Default private final long recordsWritten = -1L; + @Builder.Default private final long filesCommitted = -1L; + @Builder.Default private final String snapshotsCommitted = null; + @Builder.Default private final String partitionsCommitted = null; } diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java index fa56a577967..c82e806c641 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java @@ -60,6 +60,16 @@ public abstract class JobStatusRetriever implements LatestFlowExecutionIdTracker public static final String EVENT_NAME_FIELD = "eventName"; public static final String NA_KEY = "NA"; + // File/blob movement writers report per-dataset copy metrics on the JobSummary event via these + // flat metadata keys; the job-status monitor persists them into the job state, and they are + // surfaced here on the JobStatistics REST surface. -1 = absent/unsupported for the counts; + // the snapshot ids and partitions are comma-separated lists, null = absent/unreported. + public static final String BYTES_WRITTEN_FIELD = "bytesCopied"; + public static final String RECORDS_WRITTEN_FIELD = "rowsCopied"; + public static final String FILES_COMMITTED_FIELD = "filesCommitted"; + public static final String SNAPSHOTS_COMMITTED_FIELD = "snapshotsCommitted"; + public static final String PARTITIONS_COMMITTED_FIELD = "partitionsCommitted"; + @Getter protected final MetricContext metricContext; @@ -156,13 +166,20 @@ public static JobStatus.JobStatusBuilder createJobStatusBuilderFromState(State j boolean shouldRetry = Boolean.parseBoolean(jobState.getProp(TimingEvent.FlowEventConstants.SHOULD_RETRY_FIELD, "false")); int progressPercentage = jobState.getPropAsInt(TimingEvent.JOB_COMPLETION_PERCENTAGE, 0); long lastProgressEventTime = jobState.getPropAsLong(TimingEvent.JOB_LAST_PROGRESS_EVENT_TIME, 0); + long bytesWritten = jobState.getPropAsLong(BYTES_WRITTEN_FIELD, -1L); + long recordsWritten = jobState.getPropAsLong(RECORDS_WRITTEN_FIELD, -1L); + long filesCommitted = jobState.getPropAsLong(FILES_COMMITTED_FIELD, -1L); + String snapshotsCommitted = jobState.getProp(SNAPSHOTS_COMMITTED_FIELD, null); + String partitionsCommitted = jobState.getProp(PARTITIONS_COMMITTED_FIELD, null); return JobStatus.builder().flowName(flowName).flowGroup(flowGroup).flowExecutionId(flowExecutionId).jobName(jobName) .jobGroup(jobGroup).jobTag(jobTag).jobExecutionId(jobExecutionId).eventName(eventName).lowWatermark(lowWatermark) .highWatermark(highWatermark).orchestratedTime(orchestratedTime).startTime(startTime).endTime(endTime) .message(message).processedCount(processedCount).maxAttempts(maxAttempts).currentAttempts(currentAttempts) .currentGeneration(currentGeneration).shouldRetry(shouldRetry).progressPercentage(progressPercentage) - .lastProgressEventTime(lastProgressEventTime); + .lastProgressEventTime(lastProgressEventTime) + .bytesWritten(bytesWritten).recordsWritten(recordsWritten).filesCommitted(filesCommitted) + .snapshotsCommitted(snapshotsCommitted).partitionsCommitted(partitionsCommitted); } protected static final String getFlowGroup(State jobState) { From 8919efdf1f0ca2d70047dbaf2591422def19020d Mon Sep 17 00:00:00 2001 From: pratapaditya04 Date: Thu, 30 Jul 2026 12:01:40 +0530 Subject: [PATCH 2/3] Drop the unrelated flow-level 'issues' field from the flowexecutions snapshot FlowExecution.pdl already declares 'issues' on master; this branch never touched that file. The snapshot json was simply stale, so regenerating it picked the field up as an incidental diff. Removing it keeps the PR scoped to the copy metrics. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- ...rg.apache.gobblin.service.flowexecutions.snapshot.json | 8 -------- 1 file changed, 8 deletions(-) diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json index 57cdd9c7c3e..ff29c500693 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json @@ -283,14 +283,6 @@ } }, "doc" : "Status of jobs belonging to the flow" - }, { - "name" : "issues", - "type" : { - "type" : "array", - "items" : "Issue" - }, - "doc" : "Collection of flow-level issues from the orchestration layer.\nPopulated when errors occur before any job starts (e.g., compilation failures)\nor when an error affects the entire flow (e.g., flow SLA exceeded).", - "default" : [ ] } ] }, "org.apache.gobblin.service.FlowId", "org.apache.gobblin.service.FlowStatistics", "org.apache.gobblin.service.FlowStatusId", "org.apache.gobblin.service.Issue", "org.apache.gobblin.service.IssueSeverity", "org.apache.gobblin.service.JobId", "org.apache.gobblin.service.JobState", "org.apache.gobblin.service.JobStatistics", "org.apache.gobblin.service.JobStatus", "org.apache.gobblin.service.Timestamp" ], "schema" : { From 4b5ab8d939e9e4f19b7b869f07617f05d72272f6 Mon Sep 17 00:00:00 2001 From: pratapaditya04 Date: Fri, 31 Jul 2026 11:36:42 +0530 Subject: [PATCH 3/3] Fix findbugs UWF_NULL_FIELD on the new DatasetTaskSummary fields The 'Run static checks' job failed on :gobblin-runtime:findbugsMain with two UWF_NULL_FIELD violations: snapshotsCommitted and partitionsCommitted were declared with an explicit '= null' initializer and no in-repo writer, so FindBugs correctly concluded every read returns null. They are normally populated reflectively during JSON deserialization, which FindBugs cannot see. Add @AllArgsConstructor to give native producers a real write path and drop the redundant '= null' initializers (fields already default to null). The 5-arg @RequiredArgsConstructor is untouched, so existing call sites still compile. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../apache/gobblin/runtime/DatasetTaskSummary.java | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java index 88a493077e5..adc8a15159e 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java @@ -18,6 +18,7 @@ package org.apache.gobblin.runtime; import lombok.AccessLevel; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import lombok.NonNull; @@ -34,6 +35,7 @@ @Data @Setter(AccessLevel.NONE) // NOTE: non-`final` members solely to enable deserialization @RequiredArgsConstructor +@AllArgsConstructor @NoArgsConstructor @ToString public class DatasetTaskSummary { @@ -43,11 +45,12 @@ public class DatasetTaskSummary { @NonNull private boolean successfullyCommitted; @NonNull private String dataQualityStatus; // NOTE: intentionally NOT @NonNull so the 5-arg @RequiredArgsConstructor stays intact for native - // Gobblin call sites (AbstractJobLauncher). Populated via reflection during JSON deserialization - // of events that carry them (e.g. DDM Iceberg snapshot replication). Both are comma-separated - // lists (a run can commit more than one snapshot/partition); null = unsupported/unreported. - private String snapshotsCommitted = null; - private String partitionsCommitted = null; + // Gobblin call sites (AbstractJobLauncher). Usually populated via reflection during JSON + // deserialization of events that carry them (e.g. DDM Iceberg snapshot replication); the + // @AllArgsConstructor gives native producers a direct path. Both are comma-separated lists (a + // run can commit more than one snapshot/partition); null = unsupported/unreported. + private String snapshotsCommitted; + private String partitionsCommitted; /** * Convert a {@link DatasetTaskSummary} to a {@link DatasetMetric}.