replace all java Collectors.toList with kotlin construct

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/StandardNameTransformer.kt

@@ -8,7 +8,6 @@ import io.airbyte.commons.json.Jsons
 import io.airbyte.commons.string.Strings
 import io.airbyte.commons.text.Names
 import io.airbyte.commons.util.MoreIterators
-import java.util.stream.Collectors
 
 open class StandardNameTransformer : NamingConventionTransformer {
     override fun getIdentifier(name: String): String {
@@ -77,7 +76,7 @@ open class StandardNameTransformer : NamingConventionTransformer {
                     MoreIterators.toList(root.elements())
                         .stream()
                         .map { r: JsonNode -> formatJsonPath(r) }
-                        .collect(Collectors.toList())
+                        .toList()
                 )
             } else {
                 return root

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/async/DetectStreamToFlush.kt

@@ -273,7 +273,7 @@ internal constructor(
                     }
                     .thenComparing { s: StreamDescriptor -> s.namespace + s.name },
             )
-            .collect(Collectors.toList())
+            .toList()
     }
 
     companion object {

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/concurrent/ConcurrentStreamConsumer.kt

@@ -13,7 +13,6 @@ import java.util.*
 import java.util.concurrent.*
 import java.util.concurrent.ThreadPoolExecutor.AbortPolicy
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import kotlin.math.min
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -85,7 +84,7 @@ class ConcurrentStreamConsumer(
                 .map { runnable: ConcurrentStreamRunnable ->
                     CompletableFuture.runAsync(runnable, executorService)
                 }
-                .collect(Collectors.toList())
+                .toList()
 
         /*
          * Wait for the submitted streams to complete before returning. This uses the join() method to allow

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestJdbcUtils.kt

@@ -20,7 +20,6 @@ import io.airbyte.commons.string.Strings
 import io.airbyte.protocol.models.JsonSchemaType
 import java.math.BigDecimal
 import java.sql.*
-import java.util.stream.Collectors
 import javax.sql.DataSource
 import org.bouncycastle.util.encoders.Base64
 import org.junit.jupiter.api.Assertions
@@ -122,7 +121,7 @@ internal class TestJdbcUtils {
                 JdbcDatabase.toUnsafeStream(rs) { queryContext: ResultSet ->
                         sourceOperations.rowToJson(queryContext)
                     }
-                    .collect(Collectors.toList())
+                    .toList()
             Assertions.assertEquals(RECORDS_AS_JSON, actual)
         }
     }

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunnerTest.kt

@@ -24,7 +24,6 @@ import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import org.apache.commons.lang3.ThreadUtils
 import org.assertj.core.api.AssertionsForClassTypes
 import org.junit.jupiter.api.Assertions
@@ -474,7 +473,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
             ThreadUtils.getAllThreads()
                 .stream()
                 .filter(IntegrationRunner::filterOrphanedThread)
-                .collect(Collectors.toList())
+                .toList()
         // all threads should be interrupted
         Assertions.assertEquals(listOf<Any>(), runningThreads)
         Assertions.assertEquals(1, caughtExceptions.size)
@@ -502,7 +501,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
             ThreadUtils.getAllThreads()
                 .stream()
                 .filter(IntegrationRunner::filterOrphanedThread)
-                .collect(Collectors.toList())
+                .toList()
         // a thread that refuses to be interrupted should remain
         Assertions.assertEquals(1, runningThreads.size)
         Assertions.assertEquals(1, caughtExceptions.size)

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/destination/async/AsyncStreamConsumerTest.kt

@@ -37,7 +37,6 @@ import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicLong
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import java.util.stream.Stream
 import org.apache.commons.lang3.RandomStringUtils
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -574,7 +573,7 @@ class AsyncStreamConsumerTest {
                             ),
                         )
                 }
-                .collect(Collectors.toList())
+                .toList()
         assertEquals(expRecords, actualRecords)
     }
 }

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/destination/buffered_stream_consumer/BufferedStreamConsumerTest.kt

@@ -17,7 +17,6 @@ import java.time.Instant
 import java.util.*
 import java.util.concurrent.TimeUnit
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import java.util.stream.Stream
 import org.apache.commons.lang3.RandomStringUtils
 import org.junit.jupiter.api.Assertions
@@ -150,7 +149,7 @@ class BufferedStreamConsumerTest {
             Lists.newArrayList(expectedRecordsBatch1, expectedRecordsBatch2)
                 .stream()
                 .flatMap { obj: List<AirbyteMessage> -> obj.stream() }
-                .collect(Collectors.toList())
+                .toList()
         verifyRecords(STREAM_NAME, SCHEMA_NAME, expectedRecords)
 
         Mockito.verify(outputRecordCollector).accept(STATE_MESSAGE1)
@@ -241,7 +240,7 @@ class BufferedStreamConsumerTest {
                 .stream()
                 .map { `object`: AirbyteMessage -> Jsons.clone(`object`) }
                 .peek { m: AirbyteMessage -> m.record.withStream(STREAM_NAME2) }
-                .collect(Collectors.toList())
+                .toList()
 
         consumer.start()
         consumeRecords(consumer, expectedRecordsStream1)
@@ -266,7 +265,7 @@ class BufferedStreamConsumerTest {
                 .stream()
                 .map { `object`: AirbyteMessage -> Jsons.clone(`object`) }
                 .peek { m: AirbyteMessage -> m.record.withStream(STREAM_NAME2) }
-                .collect(Collectors.toList())
+                .toList()
 
         consumer.start()
         consumeRecords(consumer, expectedRecordsStream1)
@@ -310,7 +309,7 @@ class BufferedStreamConsumerTest {
             STREAM_NAME,
             SCHEMA_NAME,
             Stream.concat(expectedRecordsStream1.stream(), expectedRecordsStream1Batch2.stream())
-                .collect(Collectors.toList())
+                .toList()
         )
         Mockito.verify(outputRecordCollector).accept(STATE_MESSAGE1)
     }
@@ -345,7 +344,7 @@ class BufferedStreamConsumerTest {
             STREAM_NAME,
             SCHEMA_NAME,
             Stream.concat(expectedRecordsStream1.stream(), expectedRecordsStream1Batch2.stream())
-                .collect(Collectors.toList())
+                .toList()
         )
         verifyRecords(STREAM_NAME, SCHEMA_NAME, expectedRecordsStream1Batch3)
         // expects two STATE messages returned since one will be flushed after periodic flushing
@@ -589,10 +588,7 @@ class BufferedStreamConsumerTest {
         Mockito.verify(recordWriter)
             .accept(
                 AirbyteStreamNameNamespacePair(streamName, namespace),
-                expectedRecords
-                    .stream()
-                    .map { obj: AirbyteMessage -> obj.record }
-                    .collect(Collectors.toList())
+                expectedRecords.stream().map { obj: AirbyteMessage -> obj.record }.toList()
             )
     }
 

airbyte-cdk/java/airbyte-cdk/datastore-bigquery/src/main/kotlin/io/airbyte/cdk/db/bigquery/BigQueryDatabase.kt

@@ -16,7 +16,6 @@ import java.io.IOException
 import java.sql.SQLException
 import java.util.*
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import java.util.stream.Stream
 import org.apache.commons.lang3.StringUtils
 import org.apache.commons.lang3.tuple.ImmutablePair
@@ -106,7 +105,7 @@ constructor(
                         .setType(StandardSQLTypeName.STRING)
                         .build()
                 }
-                .collect(Collectors.toList())
+                .toList()
 
         return query(sql, parameterValueList)
     }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/JdbcBufferedConsumerFactory.kt

@@ -33,7 +33,6 @@ import java.util.*
 import java.util.concurrent.Executors
 import java.util.function.Consumer
 import java.util.function.Function
-import java.util.stream.Collectors
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
@@ -111,15 +110,13 @@ object JdbcBufferedConsumerFactory {
         return if (parsedCatalog == null) {
             catalog!!
                 .streams
-                .stream()
-                .map(toWriteConfig(namingResolver, config, schemaRequired))
-                .collect(Collectors.toList())
+                .map { toWriteConfig(namingResolver, config, schemaRequired).apply(it) }
+                .toList()
         } else {
             // we should switch this to kotlin-style list processing, but meh for now
             parsedCatalog.streams
-                .stream()
-                .map(parsedStreamToWriteConfig(namingResolver))
-                .collect(Collectors.toList())
+                .map { parsedStreamToWriteConfig(namingResolver).apply(it) }
+                .toList()
         }
     }
 

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcSqlGenerator.kt

@@ -23,9 +23,10 @@ import io.airbyte.integrations.base.destination.typing_deduping.UnsupportedOneOf
 import io.airbyte.protocol.models.v0.DestinationSyncMode
 import java.sql.Timestamp
 import java.time.Instant
-import java.util.Locale
-import java.util.Optional
-import java.util.stream.Collectors
+import java.util.*
+import kotlin.Any
+import kotlin.Boolean
+import kotlin.IllegalArgumentException
 import kotlin.Int
 import org.jooq.Condition
 import org.jooq.DSLContext
@@ -176,14 +177,14 @@ constructor(
                 .map { metaColumn: Map.Entry<String?, DataType<*>?> ->
                     DSL.field(DSL.quotedName(metaColumn.key), metaColumn.value)
                 }
-                .collect(Collectors.toList())
+                .toList()
         val dataFields =
             columns.entries
                 .stream()
                 .map { column: Map.Entry<ColumnId?, AirbyteType> ->
                     DSL.field(DSL.quotedName(column.key!!.name), toDialectType(column.value))
                 }
-                .collect(Collectors.toList())
+                .toList()
         dataFields.addAll(fields)
         return dataFields
     }
@@ -227,7 +228,7 @@ constructor(
                 .map { metaColumn: Map.Entry<String?, DataType<*>?> ->
                     DSL.field(DSL.quotedName(metaColumn.key), metaColumn.value)
                 }
-                .collect(Collectors.toList())
+                .toList()
         // Use originalName with non-sanitized characters when extracting data from _airbyte_data
         val dataFields = extractRawDataFields(columns, useExpensiveSaferCasting)
         dataFields.addAll(fields)

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/staging/SerialStagingConsumerFactory.kt

@@ -22,7 +22,6 @@ import java.time.Instant
 import java.util.UUID
 import java.util.function.Consumer
 import java.util.function.Function
-import java.util.stream.Collectors
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
@@ -125,7 +124,7 @@ open class SerialStagingConsumerFactory {
             return catalog.streams
                 .stream()
                 .map(toWriteConfig(namingResolver, config, parsedCatalog, useDestinationsV2Columns))
-                .collect(Collectors.toList())
+                .toList()
         }
 
         private fun toWriteConfig(

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt

@@ -62,7 +62,6 @@ import java.util.*
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.function.Consumer
-import java.util.stream.Collectors
 import java.util.stream.Stream
 import kotlin.test.assertNotNull
 import org.junit.jupiter.api.*
@@ -486,7 +485,7 @@ abstract class DestinationAcceptanceTest {
                             else message.toString()
                         }
                 )
-                .collect(Collectors.toList())
+                .toList()
 
         val config = getConfig()
         runSyncAndVerifyStateOutput(config, largeNumberRecords, configuredCatalog, false)
@@ -861,7 +860,7 @@ abstract class DestinationAcceptanceTest {
                     }
                     message
                 }
-                .collect(Collectors.toList())
+                .toList()
         assertSameMessages(expectedMessages, actualMessages, true)
     }
 
@@ -1031,7 +1030,7 @@ abstract class DestinationAcceptanceTest {
                             it.record.data["NZD"].asText()
                     (it.record.emittedAt == latestMessagesOnly[key]!!.record.emittedAt)
                 }
-                .collect(Collectors.toList())
+                .toList()
 
         val defaultSchema = getDefaultSchema(config)
         retrieveRawRecordsAndAssertSameMessages(
@@ -1754,7 +1753,7 @@ abstract class DestinationAcceptanceTest {
                     if (pruneAirbyteInternalFields) safePrune(recordMessage) else recordMessage
                 }
                 .map { obj: AirbyteRecordMessage -> obj.data }
-                .collect(Collectors.toList())
+                .toList()
 
         val actualProcessed =
             actual
@@ -1763,7 +1762,7 @@ abstract class DestinationAcceptanceTest {
                     if (pruneAirbyteInternalFields) safePrune(recordMessage) else recordMessage
                 }
                 .map { obj: AirbyteRecordMessage -> obj.data }
-                .collect(Collectors.toList())
+                .toList()
 
         _testDataComparator.assertSameData(expectedProcessed, actualProcessed)
     }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/jdbc/AbstractJdbcSource.kt

@@ -771,7 +771,7 @@ abstract class AbstractJdbcSource<Datatype>(
                 )
             }
             .map { `object`: ConfiguredAirbyteStream -> Jsons.clone(`object`) }
-            .collect(Collectors.toList())
+            .toList()
     }
 
     companion object {

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/AbstractDbSource.kt

@@ -176,7 +176,7 @@ protected constructor(driverClassName: String) :
         val iteratorList =
             Stream.of(incrementalIterators, fullRefreshIterators)
                 .flatMap(Collection<AutoCloseableIterator<AirbyteMessage>>::stream)
-                .collect(Collectors.toList())
+                .toList()
 
         return AutoCloseableIterators.appendOnClose(
             AutoCloseableIterators.concatWithEagerClose(
@@ -318,7 +318,7 @@ protected constructor(driverClassName: String) :
                 .filter { table: TableInfo<CommonField<DataType>> ->
                     !systemNameSpaces.contains(table.nameSpace) && !systemViews.contains(table.name)
                 }
-                .collect(Collectors.toList()))
+                .toList())
     }
 
     protected fun getFullRefreshIterators(
@@ -434,7 +434,7 @@ protected constructor(driverClassName: String) :
                 .stream()
                 .map { obj: CommonField<DataType> -> obj.name }
                 .filter { o: String -> selectedFieldsInCatalog.contains(o) }
-                .collect(Collectors.toList())
+                .toList()
 
         val iterator: AutoCloseableIterator<AirbyteMessage>
         // checks for which sync mode we're using based on the configured airbytestream