remove unnecessary calls to toList() (#37540)

## What
<!--
* Describe what the change is solving. Link all GitHub issues related to this change.
-->

## How
<!--
* Describe how code changes achieve the solution.
-->

## Review guide
<!--
1. `x.py`
2. `y.py`
-->

## User Impact
<!--
* What is the end result perceived by the user?
* If there are negative side effects, please list them. 
-->

## Can this PR be safely reverted and rolled back?
<!--
* If unsure, leave it blank.
-->
- [ ] YES 💚
- [ ] NO ❌

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunner.kt

@@ -481,7 +481,7 @@ internal constructor(
         ) {
             val currentThread = Thread.currentThread()
 
-            val runningThreads = ThreadUtils.getAllThreads().filter(::filterOrphanedThread).toList()
+            val runningThreads = ThreadUtils.getAllThreads().filter(::filterOrphanedThread)
             if (runningThreads.isNotEmpty()) {
                 LOGGER.warn(
                     """

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

@@ -73,9 +73,7 @@ open class StandardNameTransformer : NamingConventionTransformer {
                 return Jsons.jsonNode<Map<String, JsonNode>>(properties)
             } else if (root.isArray) {
                 return Jsons.jsonNode(
-                    MoreIterators.toList(root.elements())
-                        .map { r: JsonNode -> formatJsonPath(r) }
-                        .toList()
+                    MoreIterators.toList(root.elements()).map { r: JsonNode -> formatJsonPath(r) }
                 )
             } else {
                 return root

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

@@ -241,20 +241,18 @@ internal constructor(
                     streamDescriptor,
                 )
             }
-        return streams
-            .sortedWith(
-                Comparator.comparing(
-                        { s: StreamDescriptor -> sdToQueueSize[s]!!.orElseThrow() },
-                        Comparator.reverseOrder(),
-                    ) // if no time is present, it suggests the queue has no records. set MAX time
-                    // as a sentinel value to
-                    // represent no records.
-                    .thenComparing { s: StreamDescriptor ->
-                        sdToTimeOfLastRecord[s]!!.orElse(Instant.MAX)
-                    }
-                    .thenComparing { s: StreamDescriptor -> s.namespace + s.name },
-            )
-            .toList()
+        return streams.sortedWith(
+            Comparator.comparing(
+                    { s: StreamDescriptor -> sdToQueueSize[s]!!.orElseThrow() },
+                    Comparator.reverseOrder(),
+                ) // if no time is present, it suggests the queue has no records. set MAX time
+                // as a sentinel value to
+                // represent no records.
+                .thenComparing { s: StreamDescriptor ->
+                    sdToTimeOfLastRecord[s]!!.orElse(Instant.MAX)
+                }
+                .thenComparing { s: StreamDescriptor -> s.namespace + s.name },
+        )
     }
 
     companion object {

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

@@ -281,9 +281,7 @@ class GlobalAsyncStateManager(private val memoryManager: GlobalMemoryManager) {
             // into the non-STREAM world for correctness.
             synchronized(lock) {
                 aliasIds.addAll(
-                    descToStateIdQ.values
-                        .flatMap { obj: LinkedBlockingDeque<Long> -> obj }
-                        .toList(),
+                    descToStateIdQ.values.flatMap { obj: LinkedBlockingDeque<Long> -> obj },
                 )
                 descToStateIdQ.clear()
                 retroactiveGlobalStateId = StateIdProvider.nextId

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

@@ -122,11 +122,7 @@ object SentryExceptionHelper {
                             errorMessageAndType[ErrorMapKeys.ERROR_MAP_MESSAGE_KEY] =
                                 String.format(
                                     "%s",
-                                    stacktraceLines[
-                                            Arrays.stream(stacktraceLines)
-                                                .toList()
-                                                .indexOf(followingLine) + 1
-                                        ]
+                                    stacktraceLines[stacktraceLines.indexOf(followingLine) + 1]
                                         .trim { it <= ' ' }
                                 )
                             errorMessageAndType[ErrorMapKeys.ERROR_MAP_TYPE_KEY] =

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

@@ -71,7 +71,7 @@ class InMemoryRecordBufferingStrategy(
             stream.name,
             streamBuffer[stream]!!.size
         )
-        recordWriter.accept(stream, streamBuffer[stream]!!.toList())
+        recordWriter.accept(stream, streamBuffer[stream]!!)
         LOGGER.info("Flushing completed for {}", stream.name)
     }
 

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

@@ -123,12 +123,12 @@ object ConnectorExceptionUtil {
         initialMessage: String,
         eithers: List<Either<out T, Result>>
     ): List<Result> {
-        val throwables: List<T> = eithers.filter { it.isLeft() }.map { it.left!! }.toList()
+        val throwables: List<T> = eithers.filter { it.isLeft() }.map { it.left!! }
         if (throwables.isNotEmpty()) {
             logAllAndThrowFirst(initialMessage, throwables)
         }
         // No need to filter on isRight since isLeft will throw before reaching this line.
-        return eithers.map { obj: Either<out T, Result> -> obj.right!! }.toList()
+        return eithers.map { obj: Either<out T, Result> -> obj.right!! }
     }
 
     private fun isTransientErrorException(e: Throwable?): Boolean {

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

@@ -83,7 +83,6 @@ class ConcurrentStreamConsumer(
                 .map { runnable: ConcurrentStreamRunnable ->
                     CompletableFuture.runAsync(runnable, executorService)
                 }
-                .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

@@ -119,10 +119,10 @@ internal class TestJdbcUtils {
             val rs = connection.createStatement().executeQuery("SELECT * FROM id_and_name;")
             val actual =
                 JdbcDatabase.toUnsafeStream(rs) { queryContext: ResultSet ->
-                        sourceOperations.rowToJson(queryContext)
-                    }
-                    .toList()
-            Assertions.assertEquals(RECORDS_AS_JSON, actual)
+                    sourceOperations.rowToJson(queryContext)
+                }
+
+            Assertions.assertEquals(RECORDS_AS_JSON, actual.toList())
         }
     }
 

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

@@ -144,7 +144,7 @@ internal class TestStreamingJdbcDatabase {
         // This check assumes that FetchSizeConstants.TARGET_BUFFER_BYTE_SIZE = 200 MB.
         // Update this check if the buffer size constant is changed.
         Assertions.assertEquals(2, fetchSizes.size)
-        val sortedSizes = fetchSizes.sorted().toList()
+        val sortedSizes = fetchSizes.sorted()
         Assertions.assertTrue(sortedSizes[0] < FetchSizeConstants.INITIAL_SAMPLE_SIZE)
         Assertions.assertEquals(FetchSizeConstants.INITIAL_SAMPLE_SIZE, sortedSizes[1])
     }

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

@@ -548,12 +548,7 @@ class AsyncStreamConsumerTest {
             )
 
         // captures the output of all the workers, since our records could come out in any of them.
-        val actualRecords =
-            argumentCaptor.allValues
-                .stream() // flatten those results into a single list for the simplicity of
-                // comparison
-                .flatMap { s: Stream<*> -> s }
-                .toList()
+        val actualRecords = argumentCaptor.allValues.flatMap { it.toList() }
 
         val expRecords =
             allRecords.map { m: AirbyteMessage ->