Skip to content

Commit a4021dc

Browse files
stephane-airbytestephane-airbyte
committed
fix compilation errors
1 parent 81bab23 commit a4021dc

File tree

7 files changed

+294
-147
lines changed

7 files changed

+294
-147
lines changed

‎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.28.4
1+
version=0.28.5

‎airbyte-cdk/java/airbyte-cdk/datastore-mongo/build.gradle

+21
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,31 @@
1+
import org.jetbrains.kotlin.gradle.dsl.JvmTarget
2+
import org.jetbrains.kotlin.gradle.dsl.KotlinVersion
3+
14
java {
25
// TODO: rewrite code to avoid javac wornings in the first place
36
compileJava {
47
options.compilerArgs += "-Xlint:-try,-unchecked"
58
}
69
}
710

11+
compileTestFixturesKotlin {
12+
compilerOptions {
13+
allWarningsAsErrors = false
14+
}
15+
}
16+
17+
compileTestKotlin {
18+
compilerOptions {
19+
allWarningsAsErrors = false
20+
}
21+
}
22+
23+
compileKotlin {
24+
compilerOptions {
25+
allWarningsAsErrors = false
26+
}
27+
}
28+
829
dependencies {
930
implementation project(':airbyte-cdk:java:airbyte-cdk:dependencies')
1031
implementation project(':airbyte-cdk:java:airbyte-cdk:core')

‎airbyte-cdk/java/airbyte-cdk/datastore-mongo/src/main/kotlin/io/airbyte/cdk/db/mongodb/MongoDatabase.kt

+57-38
Original file line numberDiff line numberDiff line change
@@ -13,22 +13,23 @@ import io.airbyte.cdk.db.AbstractDatabase
1313
import io.airbyte.commons.exceptions.ConnectionErrorException
1414
import io.airbyte.commons.functional.CheckedFunction
1515
import io.airbyte.commons.util.MoreIterators
16-
import org.bson.BsonDocument
17-
import org.bson.Document
18-
import org.bson.conversions.Bson
19-
import org.slf4j.Logger
20-
import org.slf4j.LoggerFactory
2116
import java.util.*
2217
import java.util.Spliterators.AbstractSpliterator
2318
import java.util.function.Consumer
2419
import java.util.stream.Collectors
2520
import java.util.stream.Stream
2621
import java.util.stream.StreamSupport
22+
import org.bson.BsonDocument
23+
import org.bson.Document
24+
import org.bson.conversions.Bson
25+
import org.slf4j.Logger
26+
import org.slf4j.LoggerFactory
2727

28-
class MongoDatabase(connectionString: String?, databaseName: String?) : AbstractDatabase(), AutoCloseable {
28+
class MongoDatabase(connectionString: String?, databaseName: String?) :
29+
AbstractDatabase(), AutoCloseable {
2930
private var connectionString: ConnectionString? = null
3031
var database: com.mongodb.client.MongoDatabase? = null
31-
private var mongoClient: MongoClient? = null
32+
private val mongoClient: MongoClient
3233

3334
init {
3435
try {
@@ -54,14 +55,15 @@ class MongoDatabase(connectionString: String?, databaseName: String?) : Abstract
5455

5556
val collectionNames: Set<String?>
5657
get() {
57-
val collectionNames = database!!.listCollectionNames() ?: return Collections.EMPTY_SET
58-
return MoreIterators.toSet(database!!.listCollectionNames().iterator()).stream()
59-
.filter { c: String? -> !c!!.startsWith(MONGO_RESERVED_COLLECTION_PREFIX) }.collect(Collectors.toSet())
58+
val collectionNames = database!!.listCollectionNames() ?: return Collections.emptySet()
59+
return MoreIterators.toSet(database!!.listCollectionNames().iterator())
60+
.stream()
61+
.filter { c: String -> !c.startsWith(MONGO_RESERVED_COLLECTION_PREFIX) }
62+
.collect(Collectors.toSet())
6063
}
6164

62-
fun getCollection(collectionName: String?): MongoCollection<Document> {
63-
return database!!.getCollection(collectionName)
64-
.withReadConcern(ReadConcern.MAJORITY)
65+
fun getCollection(collectionName: String): MongoCollection<Document> {
66+
return database!!.getCollection(collectionName).withReadConcern(ReadConcern.MAJORITY)
6567
}
6668

6769
fun getOrCreateNewCollection(collectionName: String): MongoCollection<Document> {
@@ -73,7 +75,7 @@ class MongoDatabase(connectionString: String?, databaseName: String?) : Abstract
7375
}
7476

7577
@VisibleForTesting
76-
fun createCollection(name: String?): MongoCollection<Document> {
78+
fun createCollection(name: String): MongoCollection<Document> {
7779
database!!.createCollection(name)
7880
return database!!.getCollection(name)
7981
}
@@ -82,40 +84,57 @@ class MongoDatabase(connectionString: String?, databaseName: String?) : Abstract
8284
val name: String
8385
get() = database!!.name
8486

85-
fun read(collectionName: String?, columnNames: List<String>, filter: Optional<Bson?>): Stream<JsonNode> {
87+
fun read(
88+
collectionName: String?,
89+
columnNames: List<String>,
90+
filter: Optional<Bson?>
91+
): Stream<JsonNode> {
8692
try {
8793
val collection = database!!.getCollection(collectionName)
88-
val cursor = collection
89-
.find(filter.orElse(BsonDocument()))
90-
.batchSize(BATCH_SIZE)
91-
.cursor()
94+
val cursor =
95+
collection.find(filter.orElse(BsonDocument())).batchSize(BATCH_SIZE).cursor()
9296

93-
return getStream(cursor, CheckedFunction { document: Document -> MongoUtils.toJsonNode(document, columnNames) })
94-
.onClose {
95-
try {
96-
cursor.close()
97-
} catch (e: Exception) {
98-
throw RuntimeException(e.message, e)
99-
}
97+
return getStream(
98+
cursor,
99+
CheckedFunction { document: Document ->
100+
MongoUtils.toJsonNode(document, columnNames)
100101
}
102+
)
103+
.onClose {
104+
try {
105+
cursor.close()
106+
} catch (e: Exception) {
107+
throw RuntimeException(e.message, e)
108+
}
109+
}
101110
} catch (e: Exception) {
102-
LOGGER.error("Exception attempting to read data from collection: {}, {}", collectionName, e.message)
111+
LOGGER.error(
112+
"Exception attempting to read data from collection: {}, {}",
113+
collectionName,
114+
e.message
115+
)
103116
throw RuntimeException(e)
104117
}
105118
}
106119

107-
private fun getStream(cursor: MongoCursor<Document>, mapper: CheckedFunction<Document, JsonNode, Exception>): Stream<JsonNode> {
108-
return StreamSupport.stream(object : AbstractSpliterator<JsonNode?>(Long.MAX_VALUE, ORDERED) {
109-
override fun tryAdvance(action: Consumer<in JsonNode>): Boolean {
110-
try {
111-
val document = cursor.tryNext() ?: return false
112-
action.accept(mapper.apply(document))
113-
return true
114-
} catch (e: Exception) {
115-
throw RuntimeException(e)
120+
private fun getStream(
121+
cursor: MongoCursor<Document>,
122+
mapper: CheckedFunction<Document, JsonNode, Exception>
123+
): Stream<JsonNode> {
124+
return StreamSupport.stream(
125+
object : AbstractSpliterator<JsonNode>(Long.MAX_VALUE, ORDERED) {
126+
override fun tryAdvance(action: Consumer<in JsonNode>): Boolean {
127+
try {
128+
val document = cursor.tryNext() ?: return false
129+
action.accept(mapper.apply(document))
130+
return true
131+
} catch (e: Exception) {
132+
throw RuntimeException(e)
133+
}
116134
}
117-
}
118-
}, false)
135+
},
136+
false
137+
)
119138
}
120139

121140
companion object {

‎airbyte-cdk/java/airbyte-cdk/datastore-mongo/src/main/kotlin/io/airbyte/cdk/db/mongodb/MongoDatabaseException.kt

+2-1
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@
33
*/
44
package io.airbyte.cdk.db.mongodb
55

6-
class MongoDatabaseException(databaseName: String?) : RuntimeException(String.format(MONGO_DATA_BASE_NOT_FOUND, databaseName)) {
6+
class MongoDatabaseException(databaseName: String?) :
7+
RuntimeException(String.format(MONGO_DATA_BASE_NOT_FOUND, databaseName)) {
78
companion object {
89
const val MONGO_DATA_BASE_NOT_FOUND: String = "Data Base with given name - %s not found."
910
}

0 commit comments

Comments
 (0)