chore: Update S3 data-lake to use the latest version of the cdk (#69229)

Author: frifriSF59

airbyte-integrations/connectors/destination-s3-data-lake/gradle.properties

@@ -1,3 +1,3 @@
 testExecutionConcurrency=-1
-cdkVersion=0.1.61
+cdkVersion=0.1.69
 JunitMethodExecutionTimeout=10m

airbyte-integrations/connectors/destination-s3-data-lake/src/main/kotlin/io/airbyte/integrations/destination/s3_data_lake/S3DataLakeStreamLoader.kt

@@ -101,7 +101,8 @@ class S3DataLakeStreamLoader(
             // In principle, this doesn't matter, but the iceberg SDK throws an error about
             // stale table metadata without this.
             table.refresh()
-            computeOrExecuteSchemaUpdate().pendingUpdate?.commit()
+            // Commit all pending schema updates in order (important for two-phase commits)
+            computeOrExecuteSchemaUpdate().pendingUpdates.forEach { it.commit() }
             table.manageSnapshots().replaceBranch(mainBranchName, stagingBranchName).commit()
 
             if (stream.isSingleGenerationTruncate()) {

airbyte-integrations/connectors/destination-s3-data-lake/src/test/kotlin/io/airbyte/integrations/destination/s3_data_lake/S3DataLakeTableSynchronizerTest.kt

@@ -97,7 +97,8 @@ class S3DataLakeTableSynchronizerTest {
             )
 
         // We expect the original schema to be returned
-        assertThat(result).isEqualTo(SchemaUpdateResult(existingSchema, pendingUpdate = null))
+        assertThat(result)
+            .isEqualTo(SchemaUpdateResult(existingSchema, pendingUpdates = emptyList()))
 
         // Verify that no calls to updateSchema() manipulation were made
         verify(exactly = 0) { mockUpdateSchema.deleteColumn(any()) }
@@ -132,7 +133,8 @@ class S3DataLakeTableSynchronizerTest {
         // The final returned schema is the table's schema after refresh
         // Since we aren't actually applying changes, just assert that it's whatever the mock
         // returns
-        assertThat(result).isEqualTo(SchemaUpdateResult(mockNewSchema, pendingUpdate = null))
+        assertThat(result)
+            .isEqualTo(SchemaUpdateResult(mockNewSchema, pendingUpdates = emptyList()))
     }
 
     @Test
@@ -406,7 +408,7 @@ class S3DataLakeTableSynchronizerTest {
 
         every { mockTable.schema() } returns existingSchema
 
-        val (schema, pendingUpdate) =
+        val (schema, pendingUpdates) =
             synchronizer.maybeApplySchemaChanges(
                 mockTable,
                 incomingSchema,
@@ -422,6 +424,6 @@ class S3DataLakeTableSynchronizerTest {
         confirmVerified(mockUpdateSchema)
 
         assertThat(schema).isSameAs(mockNewSchema)
-        assertThat(pendingUpdate).isNotNull
+        assertThat(pendingUpdates).hasSize(1)
     }
 }