Skip to content

Commit f268be0

Browse files
gisripagisripa
authored
CDK s3-destinations: fixes for s3 connector compilation (#36868)
1 parent a9238d9 commit f268be0

File tree

18 files changed

+72
-74
lines changed

18 files changed

+72
-74
lines changed

‎airbyte-cdk/java/airbyte-cdk/README.md

+1
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,7 @@ Maven and Gradle will automatically reference the correct (pinned) version of th
144144

145145
| Version | Date | Pull Request | Subject |
146146
|:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
147+
| 0.29.8 | 2024-04-08 | [\#36868](https://github.com/airbytehq/airbyte/pull/36868) | Destinations: s3-destinations Compilation fixes for connector |
147148
| 0.29.7 | 2024-04-08 | [\#36768](https://github.com/airbytehq/airbyte/pull/36768) | Destinations: Make destination state fetch/commit logic more resilient to errors |
148149
| 0.29.6 | 2024-04-05 | [\#36577](https://github.com/airbytehq/airbyte/pull/36577) | Do not send system_error trace message for config exceptions. |
149150
| 0.29.5 | 2024-04-05 | [\#36620](https://github.com/airbytehq/airbyte/pull/36620) | Missed changes - open for extension for destination-postgres |

‎airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/adaptive/AdaptiveDestinationRunner.kt

+3-2
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,13 @@ object AdaptiveDestinationRunner {
2020
private const val DEPLOYMENT_MODE_KEY = EnvVariableFeatureFlags.DEPLOYMENT_MODE
2121
private const val CLOUD_MODE = "CLOUD"
2222

23+
@JvmStatic
2324
fun baseOnEnv(): OssDestinationBuilder {
2425
val mode = System.getenv(DEPLOYMENT_MODE_KEY)
2526
return OssDestinationBuilder(mode)
2627
}
2728

28-
class OssDestinationBuilder(private val deploymentMode: String) {
29+
class OssDestinationBuilder(private val deploymentMode: String?) {
2930
fun <OT : Destination> withOssDestination(
3031
ossDestinationSupplier: Supplier<OT>
3132
): CloudDestinationBuilder<OT> {
@@ -34,7 +35,7 @@ object AdaptiveDestinationRunner {
3435
}
3536

3637
class CloudDestinationBuilder<OT : Destination>(
37-
private val deploymentMode: String,
38+
private val deploymentMode: String?,
3839
private val ossDestinationSupplier: Supplier<OT>
3940
) {
4041
fun <CT : Destination> withCloudDestination(

‎airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
version=0.29.7
1+
version=0.29.8

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

+16-26
Original file line numberDiff line numberDiff line change
@@ -38,10 +38,9 @@ object GeneralStagingFunctions {
3838
typerDeduper: TyperDeduper
3939
): OnStartFunction {
4040
return OnStartFunction {
41-
log.info(
42-
"Preparing raw tables in destination started for {} streams",
43-
writeConfigs.size
44-
)
41+
log.info {
42+
"Preparing raw tables in destination started for ${writeConfigs.size} streams"
43+
}
4544
typerDeduper.prepareSchemasAndRunMigrations()
4645

4746
// Create raw tables
@@ -53,20 +52,16 @@ object GeneralStagingFunctions {
5352
val stageName = stagingOperations.getStageName(schema, dstTableName)
5453
val stagingPath =
5554
stagingOperations.getStagingPath(
56-
SerialStagingConsumerFactory.Companion.RANDOM_CONNECTION_ID,
55+
RANDOM_CONNECTION_ID,
5756
schema,
5857
stream,
5958
writeConfig.outputTableName,
6059
writeConfig.writeDatetime
6160
)
6261

63-
log.info(
64-
"Preparing staging area in destination started for schema {} stream {}: target table: {}, stage: {}",
65-
schema,
66-
stream,
67-
dstTableName,
68-
stagingPath
69-
)
62+
log.info {
63+
"Preparing staging area in destination started for schema $schema stream $stream: target table: $dstTableName, stage: $stagingPath"
64+
}
7065

7166
stagingOperations.createSchemaIfNotExists(database, schema)
7267
stagingOperations.createTableIfNotExists(database, schema, dstTableName)
@@ -84,16 +79,14 @@ object GeneralStagingFunctions {
8479
"Unrecognized sync mode: " + writeConfig.syncMode
8580
)
8681
}
87-
log.info(
88-
"Preparing staging area in destination completed for schema {} stream {}",
89-
schema,
90-
stream
91-
)
82+
log.info {
83+
"Preparing staging area in destination completed for schema $schema stream $stream"
84+
}
9285
}
9386

9487
typerDeduper.prepareFinalTables()
9588

96-
log.info("Executing finalization of tables.")
89+
log.info { "Executing finalization of tables." }
9790
stagingOperations.executeTransaction(database, queryList)
9891
}
9992
}
@@ -167,7 +160,7 @@ object GeneralStagingFunctions {
167160
// After moving data from staging area to the target table (airybte_raw) clean up the
168161
// staging
169162
// area (if user configured)
170-
log.info("Cleaning up destination started for {} streams", writeConfigs.size)
163+
log.info { "Cleaning up destination started for ${writeConfigs.size} streams" }
171164
typerDeduper.typeAndDedupe(streamSyncSummaries)
172165
for (writeConfig in writeConfigs) {
173166
val schemaName = writeConfig.outputSchemaName
@@ -182,12 +175,9 @@ object GeneralStagingFunctions {
182175
writeConfig.outputTableName,
183176
writeConfig.writeDatetime
184177
)
185-
log.info(
186-
"Cleaning stage in destination started for stream {}. schema {}, stage: {}",
187-
writeConfig.streamName,
188-
schemaName,
189-
stagePath
190-
)
178+
log.info {
179+
"Cleaning stage in destination started for stream ${writeConfig.streamName}. schema $schemaName, stage: $stagePath"
180+
}
191181
// TODO: This is another weird manifestation of Redshift vs Snowflake using
192182
// either or variables from
193183
// stageName/StagingPath.
@@ -196,7 +186,7 @@ object GeneralStagingFunctions {
196186
}
197187
typerDeduper.commitFinalTables()
198188
typerDeduper.cleanup()
199-
log.info("Cleaning up destination completed.")
189+
log.info { "Cleaning up destination completed." }
200190
}
201191
}
202192
}

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

+6-1
Original file line numberDiff line numberDiff line change
@@ -429,6 +429,7 @@ abstract class DestinationAcceptanceTest {
429429
val configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog)
430430
val messages: List<io.airbyte.protocol.models.v0.AirbyteMessage> =
431431
MoreResources.readResource(messagesFilename)
432+
.trim()
432433
.lines()
433434
.map {
434435
Jsons.deserialize(it, io.airbyte.protocol.models.v0.AirbyteMessage::class.java)
@@ -458,6 +459,7 @@ abstract class DestinationAcceptanceTest {
458459
val configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog)
459460
val messages: List<io.airbyte.protocol.models.v0.AirbyteMessage> =
460461
MoreResources.readResource(messagesFilename)
462+
.trim()
461463
.lines()
462464
.map {
463465
Jsons.deserialize(it, io.airbyte.protocol.models.v0.AirbyteMessage::class.java)
@@ -515,6 +517,7 @@ abstract class DestinationAcceptanceTest {
515517
getProtocolVersion()
516518
)
517519
)
520+
.trim()
518521
.lines()
519522
.map {
520523
Jsons.deserialize<io.airbyte.protocol.models.v0.AirbyteMessage>(
@@ -712,6 +715,7 @@ abstract class DestinationAcceptanceTest {
712715
getProtocolVersion()
713716
)
714717
)
718+
.trim()
715719
.lines()
716720
.map { Jsons.deserialize(it, AirbyteMessage::class.java) }
717721
.toList()
@@ -1406,6 +1410,7 @@ abstract class DestinationAcceptanceTest {
14061410
getProtocolVersion()
14071411
)
14081412
)
1413+
.trim()
14091414
.lines()
14101415
.map { Jsons.deserialize(it, AirbyteMessage::class.java) }
14111416
val config = getConfig()
@@ -2328,7 +2333,7 @@ abstract class DestinationAcceptanceTest {
23282333
private fun readMessagesFromFile(
23292334
messagesFilename: String
23302335
): List<io.airbyte.protocol.models.v0.AirbyteMessage> {
2331-
return MoreResources.readResource(messagesFilename).lines().map {
2336+
return MoreResources.readResource(messagesFilename).trim().lines().map {
23322337
Jsons.deserialize(it, AirbyteMessage::class.java)
23332338
}
23342339
}

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

+1
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ class DataArgumentsProvider : ArgumentsProvider {
4343
}
4444

4545
companion object {
46+
@JvmField
4647
val EXCHANGE_RATE_CONFIG: CatalogMessageTestConfigPair =
4748
CatalogMessageTestConfigPair("exchange_rate_catalog.json", "exchange_rate_messages.txt")
4849
val EDGE_CASE_CONFIG: CatalogMessageTestConfigPair =

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/BaseS3Destination.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ protected constructor(
2929
override fun check(config: JsonNode): AirbyteConnectionStatus? {
3030
try {
3131
val destinationConfig = configFactory.getS3DestinationConfig(config, storageProvider())
32-
val s3Client = destinationConfig!!.getS3Client()
32+
val s3Client = destinationConfig.getS3Client()
3333

3434
S3BaseChecks.testIAMUserHasListObjectPermission(s3Client, destinationConfig.bucketName)
3535
S3BaseChecks.testSingleUpload(
@@ -64,7 +64,7 @@ protected constructor(
6464
return S3ConsumerFactory()
6565
.create(
6666
outputRecordCollector,
67-
S3StorageOperations(nameTransformer, s3Config!!.getS3Client(), s3Config),
67+
S3StorageOperations(nameTransformer, s3Config.getS3Client(), s3Config),
6868
nameTransformer,
6969
getCreateFunction(
7070
s3Config,

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/BlobStorageOperations.kt

+3-3
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ abstract class BlobStorageOperations protected constructor() {
1212
protected val blobDecorators: MutableList<BlobDecorator> = ArrayList()
1313

1414
abstract fun getBucketObjectPath(
15-
namespace: String,
15+
namespace: String?,
1616
streamName: String,
1717
writeDatetime: DateTime,
1818
customFormat: String
@@ -29,7 +29,7 @@ abstract class BlobStorageOperations protected constructor() {
2929
@Throws(Exception::class)
3030
abstract fun uploadRecordsToBucket(
3131
recordsData: SerializableBuffer,
32-
namespace: String,
32+
namespace: String?,
3333
objectPath: String
3434
): String?
3535

@@ -46,7 +46,7 @@ abstract class BlobStorageOperations protected constructor() {
4646
* @param pathFormat formatted string for the path
4747
*/
4848
abstract fun cleanUpBucketObject(
49-
namespace: String,
49+
namespace: String?,
5050
streamName: String,
5151
objectPath: String,
5252
pathFormat: String

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseChecks.kt

+1
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@ object S3BaseChecks {
9191
*
9292
* @param endpoint URL string representing an accessible S3 bucket
9393
*/
94+
@JvmStatic
9495
fun testCustomEndpointSecured(endpoint: String?): Boolean {
9596
// if user does not use a custom endpoint, do not fail
9697
return if (endpoint == null || endpoint.length == 0) {

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/S3ConsumerFactory.kt

+3-3
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ class S3ConsumerFactory {
6969
pathFormat
7070
)
7171
storageOperations.cleanUpBucketObject(
72-
namespace!!,
72+
namespace,
7373
stream,
7474
outputBucketPath,
7575
pathFormat
@@ -124,7 +124,7 @@ class S3ConsumerFactory {
124124
writeConfig!!.addStoredFile(
125125
storageOperations.uploadRecordsToBucket(
126126
writer,
127-
writeConfig.namespace!!,
127+
writeConfig.namespace,
128128
writeConfig.fullOutputPath
129129
)!!
130130
)
@@ -183,7 +183,7 @@ class S3ConsumerFactory {
183183
"Undefined destination sync mode"
184184
)
185185
val abStream = stream.stream
186-
val namespace = abStream.namespace
186+
val namespace: String? = abStream.namespace
187187
val streamName = abStream.name
188188
val bucketPath = s3Config.bucketPath
189189
val customOutputFormat = java.lang.String.join("/", bucketPath, s3Config.pathFormat)

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/S3DestinationConfigFactory.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@ package io.airbyte.cdk.integrations.destination.s3
66
import com.fasterxml.jackson.databind.JsonNode
77
import javax.annotation.Nonnull
88

9-
class S3DestinationConfigFactory {
10-
fun getS3DestinationConfig(
9+
open class S3DestinationConfigFactory {
10+
open fun getS3DestinationConfig(
1111
config: JsonNode,
1212
@Nonnull storageProvider: StorageProvider
1313
): S3DestinationConfig {

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/S3StorageOperations.kt

+5-7
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,6 @@ import java.util.concurrent.ConcurrentMap
3030
import java.util.concurrent.atomic.AtomicInteger
3131
import java.util.regex.Pattern
3232
import org.apache.commons.io.FilenameUtils
33-
import org.apache.commons.lang3.StringUtils
34-
import org.apache.logging.log4j.util.Strings
3533
import org.joda.time.DateTime
3634

3735
private val logger = KotlinLogging.logger {}
@@ -46,13 +44,13 @@ open class S3StorageOperations(
4644
private val partCounts: ConcurrentMap<String, AtomicInteger> = ConcurrentHashMap()
4745

4846
override fun getBucketObjectPath(
49-
namespace: String,
47+
namespace: String?,
5048
streamName: String,
5149
writeDatetime: DateTime,
5250
customFormat: String
5351
): String {
5452
val namespaceStr: String =
55-
nameTransformer.getNamespace(if (Strings.isNotBlank(namespace)) namespace else "")
53+
nameTransformer.getNamespace(if (!namespace.isNullOrBlank()) namespace else "")
5654
val streamNameStr: String = nameTransformer.getIdentifier(streamName)
5755
return nameTransformer.applyDefaultCase(
5856
customFormat
@@ -114,7 +112,7 @@ open class S3StorageOperations(
114112

115113
override fun uploadRecordsToBucket(
116114
recordsData: SerializableBuffer,
117-
namespace: String,
115+
namespace: String?,
118116
objectPath: String
119117
): String {
120118
val exceptionsThrown: MutableList<Exception> = ArrayList()
@@ -174,7 +172,7 @@ open class S3StorageOperations(
174172
val partId: String = getPartId(objectPath)
175173
val fileExtension: String = getExtension(recordsData.filename)
176174
val fullObjectKey: String =
177-
if (StringUtils.isNotBlank(s3Config.fileNamePattern)) {
175+
if (!s3Config.fileNamePattern.isNullOrBlank()) {
178176
s3FilenameTemplateManager.applyPatternToFilename(
179177
S3FilenameTemplateParameterObject.builder()
180178
.partId(partId)
@@ -291,7 +289,7 @@ open class S3StorageOperations(
291289
}
292290

293291
override fun cleanUpBucketObject(
294-
namespace: String,
292+
namespace: String?,
295293
streamName: String,
296294
objectPath: String,
297295
pathFormat: String

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/destination/s3/S3AvroParquetDestinationAcceptanceTest.kt

+1
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@ abstract class S3AvroParquetDestinationAcceptanceTest protected constructor(s3Fo
108108
@Throws(IOException::class)
109109
private fun readMessagesFromFile(messagesFilename: String): List<AirbyteMessage> {
110110
return MoreResources.readResource(messagesFilename)
111+
.trim()
111112
.lines()
112113
.map { record -> Jsons.deserialize(record, AirbyteMessage::class.java) }
113114
.toList()

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseCsvDestinationAcceptanceTest.kt

+9-6
Original file line numberDiff line numberDiff line change
@@ -45,12 +45,15 @@ abstract class S3BaseCsvDestinationAcceptanceTest : S3DestinationAcceptanceTest(
4545
val fieldTypes = getFieldTypes(streamSchema)
4646
val jsonRecords: MutableList<JsonNode> = LinkedList()
4747

48-
for (objectSummary in objectSummaries!!) {
49-
s3Client!!.getObject(objectSummary!!.bucketName, objectSummary.key).use { `object` ->
48+
for (objectSummary in objectSummaries) {
49+
s3Client!!.getObject(objectSummary.bucketName, objectSummary.key).use { `object` ->
5050
getReader(`object`).use { `in` ->
5151
val records: Iterable<CSVRecord> =
52-
CSVFormat.DEFAULT.withQuoteMode(QuoteMode.NON_NUMERIC)
53-
.withFirstRecordAsHeader()
52+
CSVFormat.Builder.create()
53+
.setHeader()
54+
.setSkipHeaderRecord(true)
55+
.setQuoteMode(QuoteMode.NON_NUMERIC)
56+
.build()
5457
.parse(`in`)
5558
StreamSupport.stream(records.spliterator(), false).forEach { r: CSVRecord ->
5659
jsonRecords.add(getJsonNode(r.toMap(), fieldTypes))
@@ -87,7 +90,7 @@ abstract class S3BaseCsvDestinationAcceptanceTest : S3DestinationAcceptanceTest(
8790
input: Map<String, String>,
8891
fieldTypes: Map<String, String>
8992
): JsonNode {
90-
val json: ObjectNode = S3DestinationAcceptanceTest.Companion.MAPPER.createObjectNode()
93+
val json: ObjectNode = MAPPER.createObjectNode()
9194

9295
if (input.containsKey(JavaBaseConstants.COLUMN_NAME_DATA)) {
9396
return Jsons.deserialize(input[JavaBaseConstants.COLUMN_NAME_DATA])
@@ -100,7 +103,7 @@ abstract class S3BaseCsvDestinationAcceptanceTest : S3DestinationAcceptanceTest(
100103
) {
101104
continue
102105
}
103-
if (value == null || value == "") {
106+
if (value == "") {
104107
continue
105108
}
106109
val type = fieldTypes[key]

‎airbyte-cdk/java/airbyte-cdk/s3-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseCsvGzipDestinationAcceptanceTest.kt

+1-2
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,13 @@ import java.io.IOException
1111
import java.io.InputStreamReader
1212
import java.io.Reader
1313
import java.nio.charset.StandardCharsets
14-
import java.util.Map
1514
import java.util.zip.GZIPInputStream
1615

1716
abstract class S3BaseCsvGzipDestinationAcceptanceTest : S3BaseCsvDestinationAcceptanceTest() {
1817
override val formatConfig: JsonNode?
1918
get() = // config without compression defaults to GZIP
2019
Jsons.jsonNode(
21-
Map.of("format_type", outputFormat, "flattening", Flattening.ROOT_LEVEL.value)
20+
mapOf("format_type" to outputFormat, "flattening" to Flattening.ROOT_LEVEL.value)
2221
)
2322

2423
@Throws(IOException::class)

0 commit comments

Comments
 (0)