remove unnecessary calls to toList()

airbytehq · Apr 24, 2024 · 5d1b68b · 5d1b68b
1 parent 7c8644f
commit 5d1b68b
Show file tree

Hide file tree

Showing 65 changed files with 403 additions and 513 deletions.
diff --git a/...va/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunner.kt b/...va/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunner.kt
@@ -422,7 +422,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(
                     """

diff --git a/...k/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/StandardNameTransformer.kt b/...k/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

diff --git a/...core/src/main/kotlin/io/airbyte/cdk/integrations/destination/async/DetectStreamToFlush.kt b/...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 {

diff --git a/...ain/kotlin/io/airbyte/cdk/integrations/destination/async/state/GlobalAsyncStateManager.kt b/...ain/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

diff --git a/...ain/kotlin/io/airbyte/cdk/integrations/destination/normalization/SentryExceptionHelper.kt b/...ain/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] =

diff --git a/.../io/airbyte/cdk/integrations/destination/record_buffer/InMemoryRecordBufferingStrategy.kt b/.../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)
     }
 

diff --git a/...rbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt b/...rbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt
@@ -93,12 +93,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 isConfigErrorException(e: Throwable?): Boolean {

diff --git a/...e/src/main/kotlin/io/airbyte/cdk/integrations/util/concurrent/ConcurrentStreamConsumer.kt b/...e/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

diff --git a/...k/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestDefaultJdbcDatabase.kt b/...k/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestDefaultJdbcDatabase.kt
@@ -82,7 +82,7 @@ internal class TestDefaultJdbcDatabase {
                 },
                 { queryContext: ResultSet -> sourceOperations.rowToJson(queryContext) }
             )
-            .use { actual -> Assertions.assertEquals(RECORDS_AS_JSON, actual.toList()) }
+            .use { actual -> Assertions.assertEquals(RECORDS_AS_JSON, actual) }
     }
 
     @Test

diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestJdbcUtils.kt b/airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestJdbcUtils.kt
@@ -119,9 +119,9 @@ internal class TestJdbcUtils {
             val rs = connection.createStatement().executeQuery("SELECT * FROM id_and_name;")
             val actual =
                 JdbcDatabase.toUnsafeStream(rs) { queryContext: ResultSet ->
-                        sourceOperations.rowToJson(queryContext)
-                    }
-                    .toList()
+                    sourceOperations.rowToJson(queryContext)
+                }
+
             Assertions.assertEquals(RECORDS_AS_JSON, actual)
         }
     }

diff --git a/...java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/db/jdbc/TestStreamingJdbcDatabase.kt b/...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])
     }

diff --git a/.../test/kotlin/io/airbyte/cdk/integrations/base/normalization/NormalizationLogParserTest.kt b/.../test/kotlin/io/airbyte/cdk/integrations/base/normalization/NormalizationLogParserTest.kt
@@ -110,16 +110,14 @@ internal class NormalizationLogParserTest {
         expectedDbtErrors: List<String>
     ) {
         val messages =
-            parser!!
-                .create(
-                    BufferedReader(
-                        InputStreamReader(
-                            ByteArrayInputStream(rawLogs.toByteArray(StandardCharsets.UTF_8)),
-                            StandardCharsets.UTF_8
-                        )
+            parser!!.create(
+                BufferedReader(
+                    InputStreamReader(
+                        ByteArrayInputStream(rawLogs.toByteArray(StandardCharsets.UTF_8)),
+                        StandardCharsets.UTF_8
                     )
                 )
-                .toList()
+            )
 
         Assertions.assertEquals(expectedMessages, messages)
         Assertions.assertEquals(expectedDbtErrors, parser!!.dbtErrors)

diff --git a/.../src/test/kotlin/io/airbyte/cdk/integrations/destination/async/AsyncStreamConsumerTest.kt b/.../src/test/kotlin/io/airbyte/cdk/integrations/destination/async/AsyncStreamConsumerTest.kt
@@ -553,7 +553,6 @@ class AsyncStreamConsumerTest {
                 .stream() // flatten those results into a single list for the simplicity of
                 // comparison
                 .flatMap { s: Stream<*> -> s }
-                .toList()
 
         val expRecords =
             allRecords.map { m: AirbyteMessage ->

diff --git a/...kotlin/io/airbyte/cdk/integrations/destination/async/state/GlobalAsyncStateManagerTest.kt b/...kotlin/io/airbyte/cdk/integrations/destination/async/state/GlobalAsyncStateManagerTest.kt
@@ -234,9 +234,9 @@ class GlobalAsyncStateManagerTest {
                     expectedDestinationStats,
                 ),
             ),
-            stateWithStats2.keys.toList(),
+            stateWithStats2.keys,
         )
-        assertEquals(listOf(expectedDestinationStats), stateWithStats2.values.toList())
+        assertEquals(listOf(expectedDestinationStats), stateWithStats2.values)
     }
 
     private fun attachDestinationStateStats(
@@ -276,9 +276,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
 
             assertThrows(
                 IllegalArgumentException::class.java,
@@ -338,9 +338,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
         }
 
         @Test
@@ -370,9 +370,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
 
             emittedStatesFromDestination.clear()
 
@@ -396,9 +396,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats2.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats2.values)
         }
 
         @Test
@@ -428,9 +428,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
             emittedStatesFromDestination.clear()
 
             stateManager.trackState(GLOBAL_STATE_MESSAGE2, STATE_MSG_SIZE, DEFAULT_NAMESPACE)
@@ -452,11 +452,11 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats2,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
             assertEquals(
                 listOf(expectedDestinationStats2),
-                stateWithStats2.values.toList(),
+                stateWithStats2.values,
             )
             emittedStatesFromDestination.clear()
 
@@ -480,9 +480,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats3.keys.toList(),
+                stateWithStats3.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats3.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats3.values)
         }
 
         @Test
@@ -515,9 +515,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
             emittedStatesFromDestination.clear()
 
             val afterConvertId0: Long = simulateIncomingRecords(STREAM1_DESC, 10, stateManager)
@@ -542,9 +542,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats2.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats2.values)
         }
     }
 
@@ -576,9 +576,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
 
             assertThrows(
                 IllegalArgumentException::class.java,
@@ -618,9 +618,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
 
             emittedStatesFromDestination.clear()
 
@@ -645,11 +645,11 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats2,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
             assertEquals(
                 listOf(expectedDestinationStats2),
-                stateWithStats2.values.toList(),
+                stateWithStats2.values,
             )
         }
 
@@ -680,9 +680,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
             emittedStatesFromDestination.clear()
 
             stateManager.trackState(STREAM1_STATE_MESSAGE2, STATE_MSG_SIZE, DEFAULT_NAMESPACE)
@@ -704,11 +704,11 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats2,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
             assertEquals(
                 listOf(expectedDestinationStats2),
-                stateWithStats2.values.toList(),
+                stateWithStats2.values,
             )
             emittedStatesFromDestination.clear()
 
@@ -733,11 +733,11 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats3,
                     ),
                 ),
-                stateWithStats3.keys.toList(),
+                stateWithStats3.keys,
             )
             assertEquals(
                 listOf(expectedDestinationStats3),
-                stateWithStats3.values.toList(),
+                stateWithStats3.values,
             )
         }
 
@@ -770,9 +770,9 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats,
                     ),
                 ),
-                stateWithStats.keys.toList(),
+                stateWithStats.keys,
             )
-            assertEquals(listOf(expectedDestinationStats), stateWithStats.values.toList())
+            assertEquals(listOf(expectedDestinationStats), stateWithStats.values)
             emittedStatesFromDestination.clear()
 
             stateManager.decrement(stream2StateId, 4)
@@ -803,11 +803,11 @@ class GlobalAsyncStateManagerTest {
                         expectedDestinationStats2,
                     ),
                 ),
-                stateWithStats2.keys.toList(),
+                stateWithStats2.keys,
             )
             assertEquals(
                 listOf(expectedDestinationStats2),
-                stateWithStats2.values.toList(),
+                stateWithStats2.values,
             )
         }
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -82,7 +82,7 @@ internal class TestDefaultJdbcDatabase { @@
                     },
                     { queryContext: ResultSet -> sourceOperations.rowToJson(queryContext) }
                 )
-                .use { actual -> Assertions.assertEquals(RECORDS_AS_JSON, actual.toList()) }
+                .use { actual -> Assertions.assertEquals(RECORDS_AS_JSON, actual) }
         }
         @Test
@@ Expand Down @@