feat: add rejected records stats to connection status endpoints (#16973)

Author: pedroslopez

airbyte-api/server-api/src/main/openapi/config.yaml

@@ -13360,6 +13360,9 @@ components:
         recordsCommitted:
           type: integer
           format: int64
+        recordsRejected:
+          type: integer
+          format: int64
         bytesEmitted:
           type: integer
           format: int64
@@ -13425,6 +13428,9 @@ components:
         bytesCommitted:
           type: integer
           format: int64
+        recordsRejected:
+          type: integer
+          format: int64
         configType:
           $ref: "#/components/schemas/JobConfigType"
     ConnectionUptimeHistoryRead:
@@ -13441,6 +13447,7 @@ components:
         - configType
         - recordsEmitted
         - recordsCommitted
+        - recordsRejected
         - bytesEmitted
         - bytesCommitted
       properties:
@@ -13470,6 +13477,9 @@ components:
         bytesCommitted:
           type: integer
           format: int64
+        recordsRejected:
+          type: integer
+          format: int64
     ConnectionSyncResultRead:
       type: object
       required:

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/ConnectionsHandler.java

@@ -1599,7 +1599,8 @@ public List<JobSyncResultRead> getConnectionDataHistory(final ConnectionDataHist
           .bytesEmitted(aggregatedStats.getBytesEmitted())
           .bytesCommitted(aggregatedStats.getBytesCommitted())
           .recordsEmitted(aggregatedStats.getRecordsEmitted())
-          .recordsCommitted(aggregatedStats.getRecordsCommitted());
+          .recordsCommitted(aggregatedStats.getRecordsCommitted())
+          .recordsRejected(aggregatedStats.getRecordsRejected());
       result.add(jobResult);
     });
 

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/JobHistoryHandler.java

@@ -415,7 +415,8 @@ public ConnectionSyncProgressRead getConnectionSyncProgress(final ConnectionIdRe
               item.recordsEmitted(streamStats.getRecordsEmitted())
                   .recordsCommitted(streamStats.getRecordsCommitted())
                   .bytesEmitted(streamStats.getBytesEmitted())
-                  .bytesCommitted(streamStats.getBytesCommitted());
+                  .bytesCommitted(streamStats.getBytesCommitted())
+                  .recordsRejected(streamStats.getRecordsRejected());
             }
 
             return item;
@@ -431,6 +432,7 @@ public ConnectionSyncProgressRead getConnectionSyncProgress(final ConnectionIdRe
         .bytesCommitted(aggregatedStats == null ? null : aggregatedStats.getBytesCommitted())
         .recordsEmitted(aggregatedStats == null ? null : aggregatedStats.getRecordsEmitted())
         .recordsCommitted(aggregatedStats == null ? null : aggregatedStats.getRecordsCommitted())
+        .recordsRejected(aggregatedStats == null ? null : aggregatedStats.getRecordsRejected())
         .configType(runningJobConfigType)
         .streams(finalStreamsWithStats);
   }

airbyte-commons-server/src/test/java/io/airbyte/commons/server/handlers/ConnectionsHandlerTest.java

@@ -2625,6 +2625,7 @@ void testGetConnectionDataHistory() throws IOException {
         final long jobTwoBytesEmmitted = 87654L;
         final long jobTwoRecordsCommitted = 50L;
         final long jobTwoRecordsEmittted = 60L;
+        final long jobTwoRecordsRejected = 10L;
         try (final MockedStatic<StatsAggregationHelper> mockStatsAggregationHelper = Mockito.mockStatic(StatsAggregationHelper.class)) {
           mockStatsAggregationHelper.when(() -> StatsAggregationHelper.getJobIdToJobWithAttemptsReadMap(Mockito.any(), Mockito.any()))
               .thenReturn(Map.of(
@@ -2641,6 +2642,7 @@ jobTwoId, new JobWithAttemptsRead().job(
                               .bytesCommitted(jobTwoBytesCommitted)
                               .bytesEmitted(jobTwoBytesEmmitted)
                               .recordsCommitted(jobTwoRecordsCommitted)
+                              .recordsRejected(jobTwoRecordsRejected)
                               .recordsEmitted(jobTwoRecordsEmittted)))));
 
           final List<JobSyncResultRead> expected = List.of(
@@ -2660,6 +2662,7 @@ jobTwoId, new JobWithAttemptsRead().job(
                   .bytesEmitted(jobTwoBytesEmmitted)
                   .recordsCommitted(jobTwoRecordsCommitted)
                   .recordsEmitted(jobTwoRecordsEmittted)
+                  .recordsRejected(jobTwoRecordsRejected)
                   .jobCreatedAt(jobTwoCreatedAt)
                   .jobUpdatedAt(jobTwoUpdatedAt));
 

airbyte-commons-server/src/test/java/io/airbyte/commons/server/handlers/JobHistoryHandlerTest.java

@@ -734,12 +734,12 @@ void testGetConnectionSyncProgressWithRunningJob() throws IOException {
           new StreamDescriptor().name("stream3")));
 
       jobRead.setStreamAggregatedStats(List.of(
-          new StreamStats().streamName("stream2").recordsEmitted(50L).bytesEmitted(20L).recordsCommitted(45L).bytesCommitted(15L),
+          new StreamStats().streamName("stream2").recordsEmitted(50L).bytesEmitted(20L).recordsCommitted(45L).bytesCommitted(15L).recordsRejected(2L),
           new StreamStats().streamName("stream1").streamNamespace("ns1").recordsEmitted(5L).bytesEmitted(2L).recordsCommitted(5L)
               .bytesCommitted(2L)));
 
       final JobAggregatedStats jobAggregatedStats =
-          new JobAggregatedStats().bytesCommitted(17L).recordsCommitted(50L).bytesEmitted(22L).recordsEmitted(55L);
+          new JobAggregatedStats().bytesCommitted(17L).recordsCommitted(50L).bytesEmitted(22L).recordsEmitted(55L).recordsRejected(2L);
       jobRead.setAggregatedStats(jobAggregatedStats);
 
       final JobWithAttemptsRead firstJobWithAttemptRead = new JobWithAttemptsRead()
@@ -755,6 +755,7 @@ void testGetConnectionSyncProgressWithRunningJob() throws IOException {
             .connectionId(connectionId)
             .bytesCommitted(jobAggregatedStats.getBytesCommitted())
             .recordsCommitted(jobAggregatedStats.getRecordsCommitted())
+            .recordsRejected(jobAggregatedStats.getRecordsRejected())
             .bytesEmitted(jobAggregatedStats.getBytesEmitted())
             .recordsEmitted(jobAggregatedStats.getRecordsEmitted())
             .configType(JobConfigType.SYNC)
@@ -774,6 +775,7 @@ void testGetConnectionSyncProgressWithRunningJob() throws IOException {
                     .bytesEmitted(20L)
                     .recordsCommitted(45L)
                     .bytesCommitted(15L)
+                    .recordsRejected(2L)
                     .configType(JobConfigType.SYNC),
                 new StreamSyncProgressReadItem()
                     .streamName("stream3")

airbyte-server/src/main/kotlin/io/airbyte/server/handlers/StreamStatusesHandler.kt

@@ -131,6 +131,7 @@ open class StreamStatusesHandler(
           .bytesCommitted(aggregatedStats.bytesCommitted)
           .recordsEmitted(aggregatedStats.recordsEmitted)
           .recordsCommitted(aggregatedStats.recordsCommitted)
+          .recordsRejected(aggregatedStats.recordsRejected)
       result.add(jobResult)
     }
 

airbyte-server/src/test/kotlin/io/airbyte/server/handlers/StreamStatusesHandlerTest.kt

@@ -202,6 +202,7 @@ internal class StreamStatusesHandlerTest {
     val jobTwoBytesEmmitted = 87654L
     val jobTwoRecordsCommitted = 50L
     val jobTwoRecordsEmittted = 60L
+    val jobTwoRecordsRejected = 10L
     Mockito.mockStatic(StatsAggregationHelper::class.java).use { mockStatsAggregationHelper ->
       mockStatsAggregationHelper
         .`when`<Any> {
@@ -228,13 +229,14 @@ internal class StreamStatusesHandlerTest {
                   .bytesCommitted(jobTwoBytesCommitted)
                   .bytesEmitted(jobTwoBytesEmmitted)
                   .recordsCommitted(jobTwoRecordsCommitted)
-                  .recordsEmitted(jobTwoRecordsEmittted),
+                  .recordsEmitted(jobTwoRecordsEmittted)
+                  .recordsRejected(jobTwoRecordsRejected),
               ),
             ),
           ),
         )
       val expected =
-        List.of(
+        listOf(
           JobSyncResultRead()
             .configType(JobConfigType.SYNC)
             .jobId(jobOneId)
@@ -245,7 +247,7 @@ internal class StreamStatusesHandlerTest {
             .jobCreatedAt(jobOneCreatedAt)
             .jobUpdatedAt(jobOneUpdatedAt)
             .streamStatuses(
-              List.of<@Valid ConnectionSyncResultRead?>(
+              listOf<@Valid ConnectionSyncResultRead?>(
                 ConnectionSyncResultRead()
                   .status(io.airbyte.api.model.generated.JobStatus.SUCCEEDED)
                   .streamName("streamOne")
@@ -263,10 +265,11 @@ internal class StreamStatusesHandlerTest {
             .bytesEmitted(jobTwoBytesEmmitted)
             .recordsCommitted(jobTwoRecordsCommitted)
             .recordsEmitted(jobTwoRecordsEmittted)
+            .recordsRejected(jobTwoRecordsRejected)
             .jobCreatedAt(jobTwoCreatedAt)
             .jobUpdatedAt(jobTwoUpdatedAt)
             .streamStatuses(
-              List.of<@Valid ConnectionSyncResultRead?>(
+              listOf<@Valid ConnectionSyncResultRead?>(
                 ConnectionSyncResultRead()
                   .status(io.airbyte.api.model.generated.JobStatus.SUCCEEDED)
                   .streamName("streamThree")

airbyte-webapp/src/area/connection/components/HistoricalOverview/HistoricalOverview.tsx

@@ -56,6 +56,7 @@ const generatePlaceholderHistory = (
       jobUpdatedAt: dayjs().unix(),
       recordsCommitted: connectionSyncProgress.recordsCommitted ?? 0,
       recordsEmitted: connectionSyncProgress.recordsEmitted ?? 0,
+      recordsRejected: connectionSyncProgress.recordsRejected ?? 0,
       streamStatuses: connectionSyncProgress.streams.map((syncProgressItem) => {
         return {
           status: "running",