Skip to content

Commit b0ab148

Browse files
stephane-airbytestephane-airbyte
authored
replace all java Collectors.toList with kotlin construct (#37537)
more kotlin cleanup
1 parent c304df3 commit b0ab148

File tree

40 files changed

+116
-157
lines changed

40 files changed

+116
-157
lines changed

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

+1-2
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import io.airbyte.commons.json.Jsons
88
import io.airbyte.commons.string.Strings
99
import io.airbyte.commons.text.Names
1010
import io.airbyte.commons.util.MoreIterators
11-
import java.util.stream.Collectors
1211

1312
open class StandardNameTransformer : NamingConventionTransformer {
1413
override fun getIdentifier(name: String): String {
@@ -77,7 +76,7 @@ open class StandardNameTransformer : NamingConventionTransformer {
7776
MoreIterators.toList(root.elements())
7877
.stream()
7978
.map { r: JsonNode -> formatJsonPath(r) }
80-
.collect(Collectors.toList())
79+
.toList()
8180
)
8281
} else {
8382
return root

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -273,7 +273,7 @@ internal constructor(
273273
}
274274
.thenComparing { s: StreamDescriptor -> s.namespace + s.name },
275275
)
276-
.collect(Collectors.toList())
276+
.toList()
277277
}
278278

279279
companion object {

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

+1-2
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,6 @@ import java.util.*
1313
import java.util.concurrent.*
1414
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy
1515
import java.util.function.Consumer
16-
import java.util.stream.Collectors
1716
import kotlin.math.min
1817
import org.slf4j.Logger
1918
import org.slf4j.LoggerFactory
@@ -85,7 +84,7 @@ class ConcurrentStreamConsumer(
8584
.map { runnable: ConcurrentStreamRunnable ->
8685
CompletableFuture.runAsync(runnable, executorService)
8786
}
88-
.collect(Collectors.toList())
87+
.toList()
8988

9089
/*
9190
* 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

+1-2
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ import io.airbyte.commons.string.Strings
2020
import io.airbyte.protocol.models.JsonSchemaType
2121
import java.math.BigDecimal
2222
import java.sql.*
23-
import java.util.stream.Collectors
2423
import javax.sql.DataSource
2524
import org.bouncycastle.util.encoders.Base64
2625
import org.junit.jupiter.api.Assertions
@@ -122,7 +121,7 @@ internal class TestJdbcUtils {
122121
JdbcDatabase.toUnsafeStream(rs) { queryContext: ResultSet ->
123122
sourceOperations.rowToJson(queryContext)
124123
}
125-
.collect(Collectors.toList())
124+
.toList()
126125
Assertions.assertEquals(RECORDS_AS_JSON, actual)
127126
}
128127
}

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

+2-3
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@ import java.util.concurrent.Executors
2424
import java.util.concurrent.TimeUnit
2525
import java.util.concurrent.atomic.AtomicBoolean
2626
import java.util.function.Consumer
27-
import java.util.stream.Collectors
2827
import org.apache.commons.lang3.ThreadUtils
2928
import org.assertj.core.api.AssertionsForClassTypes
3029
import org.junit.jupiter.api.Assertions
@@ -474,7 +473,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
474473
ThreadUtils.getAllThreads()
475474
.stream()
476475
.filter(IntegrationRunner::filterOrphanedThread)
477-
.collect(Collectors.toList())
476+
.toList()
478477
// all threads should be interrupted
479478
Assertions.assertEquals(listOf<Any>(), runningThreads)
480479
Assertions.assertEquals(1, caughtExceptions.size)
@@ -502,7 +501,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
502501
ThreadUtils.getAllThreads()
503502
.stream()
504503
.filter(IntegrationRunner::filterOrphanedThread)
505-
.collect(Collectors.toList())
504+
.toList()
506505
// a thread that refuses to be interrupted should remain
507506
Assertions.assertEquals(1, runningThreads.size)
508507
Assertions.assertEquals(1, caughtExceptions.size)

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

+1-2
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@ import java.util.concurrent.TimeUnit
3737
import java.util.concurrent.TimeoutException
3838
import java.util.concurrent.atomic.AtomicLong
3939
import java.util.function.Consumer
40-
import java.util.stream.Collectors
4140
import java.util.stream.Stream
4241
import org.apache.commons.lang3.RandomStringUtils
4342
import org.junit.jupiter.api.Assertions.assertEquals
@@ -574,7 +573,7 @@ class AsyncStreamConsumerTest {
574573
),
575574
)
576575
}
577-
.collect(Collectors.toList())
576+
.toList()
578577
assertEquals(expRecords, actualRecords)
579578
}
580579
}

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

+6-10
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ import java.time.Instant
1717
import java.util.*
1818
import java.util.concurrent.TimeUnit
1919
import java.util.function.Consumer
20-
import java.util.stream.Collectors
2120
import java.util.stream.Stream
2221
import org.apache.commons.lang3.RandomStringUtils
2322
import org.junit.jupiter.api.Assertions
@@ -150,7 +149,7 @@ class BufferedStreamConsumerTest {
150149
Lists.newArrayList(expectedRecordsBatch1, expectedRecordsBatch2)
151150
.stream()
152151
.flatMap { obj: List<AirbyteMessage> -> obj.stream() }
153-
.collect(Collectors.toList())
152+
.toList()
154153
verifyRecords(STREAM_NAME, SCHEMA_NAME, expectedRecords)
155154

156155
Mockito.verify(outputRecordCollector).accept(STATE_MESSAGE1)
@@ -241,7 +240,7 @@ class BufferedStreamConsumerTest {
241240
.stream()
242241
.map { `object`: AirbyteMessage -> Jsons.clone(`object`) }
243242
.peek { m: AirbyteMessage -> m.record.withStream(STREAM_NAME2) }
244-
.collect(Collectors.toList())
243+
.toList()
245244

246245
consumer.start()
247246
consumeRecords(consumer, expectedRecordsStream1)
@@ -266,7 +265,7 @@ class BufferedStreamConsumerTest {
266265
.stream()
267266
.map { `object`: AirbyteMessage -> Jsons.clone(`object`) }
268267
.peek { m: AirbyteMessage -> m.record.withStream(STREAM_NAME2) }
269-
.collect(Collectors.toList())
268+
.toList()
270269

271270
consumer.start()
272271
consumeRecords(consumer, expectedRecordsStream1)
@@ -310,7 +309,7 @@ class BufferedStreamConsumerTest {
310309
STREAM_NAME,
311310
SCHEMA_NAME,
312311
Stream.concat(expectedRecordsStream1.stream(), expectedRecordsStream1Batch2.stream())
313-
.collect(Collectors.toList())
312+
.toList()
314313
)
315314
Mockito.verify(outputRecordCollector).accept(STATE_MESSAGE1)
316315
}
@@ -345,7 +344,7 @@ class BufferedStreamConsumerTest {
345344
STREAM_NAME,
346345
SCHEMA_NAME,
347346
Stream.concat(expectedRecordsStream1.stream(), expectedRecordsStream1Batch2.stream())
348-
.collect(Collectors.toList())
347+
.toList()
349348
)
350349
verifyRecords(STREAM_NAME, SCHEMA_NAME, expectedRecordsStream1Batch3)
351350
// expects two STATE messages returned since one will be flushed after periodic flushing
@@ -589,10 +588,7 @@ class BufferedStreamConsumerTest {
589588
Mockito.verify(recordWriter)
590589
.accept(
591590
AirbyteStreamNameNamespacePair(streamName, namespace),
592-
expectedRecords
593-
.stream()
594-
.map { obj: AirbyteMessage -> obj.record }
595-
.collect(Collectors.toList())
591+
expectedRecords.stream().map { obj: AirbyteMessage -> obj.record }.toList()
596592
)
597593
}
598594

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

+1-2
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ import java.io.IOException
1616
import java.sql.SQLException
1717
import java.util.*
1818
import java.util.function.Consumer
19-
import java.util.stream.Collectors
2019
import java.util.stream.Stream
2120
import org.apache.commons.lang3.StringUtils
2221
import org.apache.commons.lang3.tuple.ImmutablePair
@@ -106,7 +105,7 @@ constructor(
106105
.setType(StandardSQLTypeName.STRING)
107106
.build()
108107
}
109-
.collect(Collectors.toList())
108+
.toList()
110109

111110
return query(sql, parameterValueList)
112111
}

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

+4-7
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@ import java.util.*
3333
import java.util.concurrent.Executors
3434
import java.util.function.Consumer
3535
import java.util.function.Function
36-
import java.util.stream.Collectors
3736
import org.slf4j.Logger
3837
import org.slf4j.LoggerFactory
3938

@@ -111,15 +110,13 @@ object JdbcBufferedConsumerFactory {
111110
return if (parsedCatalog == null) {
112111
catalog!!
113112
.streams
114-
.stream()
115-
.map(toWriteConfig(namingResolver, config, schemaRequired))
116-
.collect(Collectors.toList())
113+
.map { toWriteConfig(namingResolver, config, schemaRequired).apply(it) }
114+
.toList()
117115
} else {
118116
// we should switch this to kotlin-style list processing, but meh for now
119117
parsedCatalog.streams
120-
.stream()
121-
.map(parsedStreamToWriteConfig(namingResolver))
122-
.collect(Collectors.toList())
118+
.map { parsedStreamToWriteConfig(namingResolver).apply(it) }
119+
.toList()
123120
}
124121
}
125122

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

+7-6
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,10 @@ import io.airbyte.integrations.base.destination.typing_deduping.UnsupportedOneOf
2323
import io.airbyte.protocol.models.v0.DestinationSyncMode
2424
import java.sql.Timestamp
2525
import java.time.Instant
26-
import java.util.Locale
27-
import java.util.Optional
28-
import java.util.stream.Collectors
26+
import java.util.*
27+
import kotlin.Any
28+
import kotlin.Boolean
29+
import kotlin.IllegalArgumentException
2930
import kotlin.Int
3031
import org.jooq.Condition
3132
import org.jooq.DSLContext
@@ -176,14 +177,14 @@ constructor(
176177
.map { metaColumn: Map.Entry<String?, DataType<*>?> ->
177178
DSL.field(DSL.quotedName(metaColumn.key), metaColumn.value)
178179
}
179-
.collect(Collectors.toList())
180+
.toList()
180181
val dataFields =
181182
columns.entries
182183
.stream()
183184
.map { column: Map.Entry<ColumnId?, AirbyteType> ->
184185
DSL.field(DSL.quotedName(column.key!!.name), toDialectType(column.value))
185186
}
186-
.collect(Collectors.toList())
187+
.toList()
187188
dataFields.addAll(fields)
188189
return dataFields
189190
}
@@ -227,7 +228,7 @@ constructor(
227228
.map { metaColumn: Map.Entry<String?, DataType<*>?> ->
228229
DSL.field(DSL.quotedName(metaColumn.key), metaColumn.value)
229230
}
230-
.collect(Collectors.toList())
231+
.toList()
231232
// Use originalName with non-sanitized characters when extracting data from _airbyte_data
232233
val dataFields = extractRawDataFields(columns, useExpensiveSaferCasting)
233234
dataFields.addAll(fields)

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

+1-2
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import java.time.Instant
2222
import java.util.UUID
2323
import java.util.function.Consumer
2424
import java.util.function.Function
25-
import java.util.stream.Collectors
2625
import org.slf4j.Logger
2726
import org.slf4j.LoggerFactory
2827

@@ -125,7 +124,7 @@ open class SerialStagingConsumerFactory {
125124
return catalog.streams
126125
.stream()
127126
.map(toWriteConfig(namingResolver, config, parsedCatalog, useDestinationsV2Columns))
128-
.collect(Collectors.toList())
127+
.toList()
129128
}
130129

131130
private fun toWriteConfig(

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

+5-6
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,6 @@ import java.util.*
6262
import java.util.concurrent.TimeUnit
6363
import java.util.concurrent.atomic.AtomicInteger
6464
import java.util.function.Consumer
65-
import java.util.stream.Collectors
6665
import java.util.stream.Stream
6766
import kotlin.test.assertNotNull
6867
import org.junit.jupiter.api.*
@@ -486,7 +485,7 @@ abstract class DestinationAcceptanceTest {
486485
else message.toString()
487486
}
488487
)
489-
.collect(Collectors.toList())
488+
.toList()
490489

491490
val config = getConfig()
492491
runSyncAndVerifyStateOutput(config, largeNumberRecords, configuredCatalog, false)
@@ -861,7 +860,7 @@ abstract class DestinationAcceptanceTest {
861860
}
862861
message
863862
}
864-
.collect(Collectors.toList())
863+
.toList()
865864
assertSameMessages(expectedMessages, actualMessages, true)
866865
}
867866

@@ -1031,7 +1030,7 @@ abstract class DestinationAcceptanceTest {
10311030
it.record.data["NZD"].asText()
10321031
(it.record.emittedAt == latestMessagesOnly[key]!!.record.emittedAt)
10331032
}
1034-
.collect(Collectors.toList())
1033+
.toList()
10351034

10361035
val defaultSchema = getDefaultSchema(config)
10371036
retrieveRawRecordsAndAssertSameMessages(
@@ -1754,7 +1753,7 @@ abstract class DestinationAcceptanceTest {
17541753
if (pruneAirbyteInternalFields) safePrune(recordMessage) else recordMessage
17551754
}
17561755
.map { obj: AirbyteRecordMessage -> obj.data }
1757-
.collect(Collectors.toList())
1756+
.toList()
17581757

17591758
val actualProcessed =
17601759
actual
@@ -1763,7 +1762,7 @@ abstract class DestinationAcceptanceTest {
17631762
if (pruneAirbyteInternalFields) safePrune(recordMessage) else recordMessage
17641763
}
17651764
.map { obj: AirbyteRecordMessage -> obj.data }
1766-
.collect(Collectors.toList())
1765+
.toList()
17671766

17681767
_testDataComparator.assertSameData(expectedProcessed, actualProcessed)
17691768
}

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -771,7 +771,7 @@ abstract class AbstractJdbcSource<Datatype>(
771771
)
772772
}
773773
.map { `object`: ConfiguredAirbyteStream -> Jsons.clone(`object`) }
774-
.collect(Collectors.toList())
774+
.toList()
775775
}
776776

777777
companion object {

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

+3-3
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,7 @@ protected constructor(driverClassName: String) :
176176
val iteratorList =
177177
Stream.of(incrementalIterators, fullRefreshIterators)
178178
.flatMap(Collection<AutoCloseableIterator<AirbyteMessage>>::stream)
179-
.collect(Collectors.toList())
179+
.toList()
180180

181181
return AutoCloseableIterators.appendOnClose(
182182
AutoCloseableIterators.concatWithEagerClose(
@@ -318,7 +318,7 @@ protected constructor(driverClassName: String) :
318318
.filter { table: TableInfo<CommonField<DataType>> ->
319319
!systemNameSpaces.contains(table.nameSpace) && !systemViews.contains(table.name)
320320
}
321-
.collect(Collectors.toList()))
321+
.toList())
322322
}
323323

324324
protected fun getFullRefreshIterators(
@@ -434,7 +434,7 @@ protected constructor(driverClassName: String) :
434434
.stream()
435435
.map { obj: CommonField<DataType> -> obj.name }
436436
.filter { o: String -> selectedFieldsInCatalog.contains(o) }
437-
.collect(Collectors.toList())
437+
.toList()
438438

439439
val iterator: AutoCloseableIterator<AirbyteMessage>
440440
// checks for which sync mode we're using based on the configured airbytestream

0 commit comments

Comments
 (0)