Skip to content

Commit 79e33fa

Browse files
davidsloandavidsloan
authored
Adding option to process without exactly-once semantics (#75)
* Adding option to process without writing an index * Renaming exactly once property
1 parent 2866a69 commit 79e33fa

20 files changed

Lines changed: 379 additions & 117 deletions

File tree

‎kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3AvroWriterManagerTest.scala‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
104104
dataStorage = DataStorageSettings.disabled,
105105
),
106106
),
107-
indexOptions = IndexOptions(5, ".indexes"),
107+
indexOptions = IndexOptions(5, ".indexes").some,
108108
compressionCodec = compressionCodec,
109109
batchDelete = true,
110110
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
@@ -116,7 +116,7 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
116116
val sink = writerManagerCreator.from(avroConfig(new OffsetFileNamer(
117117
identity[String],
118118
AvroFormatSelection.extension,
119-
)))
119+
)))._2
120120
firstUsers.zipWithIndex.foreach {
121121
case (struct: Struct, index: Int) =>
122122
val writeRes = sink.write(
@@ -152,7 +152,7 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
152152
val sink = writerManagerCreator.from(avroConfig(new OffsetFileNamer(
153153
identity[String],
154154
AvroFormatSelection.extension,
155-
)))
155+
)))._2
156156
firstUsers.zip(List(0 -> 100, 1 -> 99, 2 -> 101, 3 -> 102)).foreach {
157157
case (struct: Struct, (index: Int, timestamp: Int)) =>
158158
val writeRes = sink.write(
@@ -187,7 +187,7 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
187187
val sink = writerManagerCreator.from(avroConfig(new OffsetFileNamer(
188188
identity[String],
189189
AvroFormatSelection.extension,
190-
)))
190+
)))._2
191191
val usersWithDecimal1 =
192192
new Struct(UsersSchemaDecimal)
193193
.put("name", "sam")
@@ -266,7 +266,7 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
266266
val sink = writerManagerCreator.from(avroConfig(new OffsetFileNamer(
267267
identity[String],
268268
AvroFormatSelection.extension,
269-
)))
269+
)))._2
270270
firstUsers.concat(usersWithNewSchema).zipWithIndex.foreach {
271271
case (user, index) =>
272272
sink.write(

‎kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3JsonWriterManagerTest.scala‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -105,15 +105,15 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
105105
dataStorage = DataStorageSettings.disabled,
106106
), // JsonS3Format
107107
),
108-
indexOptions = IndexOptions(5, ".indexes"),
108+
indexOptions = IndexOptions(5, ".indexes").some,
109109
compressionCodec,
110110
batchDelete = true,
111111
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
112112
connectorRetryConfig = new RetryConfig(1, 1L, 1.0),
113113
logMetrics = false,
114114
)
115115

116-
val sink = writerManagerCreator.from(config)
116+
val sink = writerManagerCreator.from(config)._2
117117
val topic = Topic(TopicName)
118118
val offset = Offset(1)
119119
sink.write(
@@ -169,15 +169,15 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
169169
dataStorage = DataStorageSettings.disabled,
170170
),
171171
),
172-
indexOptions = IndexOptions(5, ".indexes"),
172+
indexOptions = IndexOptions(5, ".indexes").some,
173173
compressionCodec,
174174
batchDelete = true,
175175
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
176176
connectorRetryConfig = new RetryConfig(1, 1L, 1.0),
177177
logMetrics = false,
178178
)
179179

180-
val sink = writerManagerCreator.from(config)
180+
val sink = writerManagerCreator.from(config)._2
181181
firstUsers.zipWithIndex.foreach {
182182
case (struct: Struct, index: Int) =>
183183
val topic = Topic(TopicName)
@@ -237,15 +237,15 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
237237
dataStorage = DataStorageSettings.disabled,
238238
),
239239
),
240-
indexOptions = IndexOptions(5, ".indexes"),
240+
indexOptions = IndexOptions(5, ".indexes").some,
241241
compressionCodec,
242242
batchDelete = true,
243243
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
244244
connectorRetryConfig = new RetryConfig(1, 1L, 1.0),
245245
logMetrics = false,
246246
)
247247

248-
val sink = writerManagerCreator.from(config)
248+
val sink = writerManagerCreator.from(config)._2
249249

250250
val topic = Topic(TopicName)
251251
val offset = Offset(1L)
@@ -311,15 +311,15 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
311311
dataStorage = DataStorageSettings.disabled,
312312
),
313313
),
314-
indexOptions = IndexOptions(5, ".indexes"),
314+
indexOptions = IndexOptions(5, ".indexes").some,
315315
compressionCodec,
316316
batchDelete = true,
317317
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
318318
connectorRetryConfig = new RetryConfig(1, 1L, 1.0),
319319
logMetrics = false,
320320
)
321321

322-
val sink = writerManagerCreator.from(config)
322+
val sink = writerManagerCreator.from(config)._2
323323
val topic = Topic(TopicName)
324324
val offset = Offset(1)
325325
val listOfPojo: java.util.List[Pojo] = List(

‎kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3ParquetWriterManagerTest.scala‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ class S3ParquetWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyC
102102
dataStorage = DataStorageSettings.disabled,
103103
),
104104
),
105-
indexOptions = IndexOptions(5, ".indexes"),
105+
indexOptions = IndexOptions(5, ".indexes").some,
106106
compressionCodec,
107107
batchDelete = true,
108108
errorPolicy = ErrorPolicy(ErrorPolicyEnum.THROW),
@@ -112,7 +112,7 @@ class S3ParquetWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyC
112112

113113
"parquet sink" should "write 2 records to parquet format in s3" in {
114114

115-
val sink = writerManagerCreator.from(parquetConfig)
115+
val sink = writerManagerCreator.from(parquetConfig)._2
116116
firstUsers.zipWithIndex.foreach {
117117
case (struct: Struct, index: Int) =>
118118
val topic = Topic(TopicName)
@@ -158,7 +158,7 @@ class S3ParquetWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyC
158158
new Struct(secondSchema).put("name", "coco").put("designation", null).put("salary", 395.44),
159159
)
160160

161-
val sink = writerManagerCreator.from(parquetConfig)
161+
val sink = writerManagerCreator.from(parquetConfig)._2
162162
firstUsers.concat(usersWithNewSchema).zipWithIndex.foreach {
163163
case (user, index) =>
164164
val topic = Topic(TopicName)

‎kafka-connect-aws-s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/WriterManagerTest.scala‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,14 @@
11
package io.lenses.streamreactor.connect.aws.s3.sink
22

33
import cats.implicits.catsSyntaxEitherId
4+
import cats.implicits.catsSyntaxOptionId
45
import io.lenses.streamreactor.connect.aws.s3.model.location.S3LocationValidator
56
import io.lenses.streamreactor.connect.aws.s3.storage.S3FileMetadata
67
import io.lenses.streamreactor.connect.aws.s3.utils.S3ProxyContainerTest
78
import io.lenses.streamreactor.connect.cloud.common.formats.writer.FormatWriter
9+
import io.lenses.streamreactor.connect.cloud.common.model.Offset
810
import io.lenses.streamreactor.connect.cloud.common.model.Topic
11+
import io.lenses.streamreactor.connect.cloud.common.model.TopicPartition
912
import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocation
1013
import io.lenses.streamreactor.connect.cloud.common.model.location.CloudLocationValidator
1114
import io.lenses.streamreactor.connect.cloud.common.sink.commit.CommitPolicy
@@ -14,7 +17,7 @@ import io.lenses.streamreactor.connect.cloud.common.sink.commit.FileSize
1417
import io.lenses.streamreactor.connect.cloud.common.sink.commit.Interval
1518
import io.lenses.streamreactor.connect.cloud.common.sink.naming.CloudKeyNamer
1619
import io.lenses.streamreactor.connect.cloud.common.sink.naming.ObjectKeyBuilder
17-
import io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManager
20+
import io.lenses.streamreactor.connect.cloud.common.sink.writer.WriterIndexer
1821
import io.lenses.streamreactor.connect.cloud.common.sink.writer.WriterManager
1922
import org.apache.kafka.clients.consumer.OffsetAndMetadata
2023
import org.mockito.MockitoSugar
@@ -39,8 +42,9 @@ class WriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyContainerT
3942
stagingFilenameFn = (_, _) => new File("blah.csv").asRight,
4043
objKeyBuilderFn = (_, _) => mock[ObjectKeyBuilder],
4144
formatWriterFn = (_, _) => mock[FormatWriter].asRight,
42-
indexManager = mock[IndexManager[S3FileMetadata]],
45+
writerIndexer = mock[WriterIndexer[S3FileMetadata]],
4346
_.asRight,
47+
((_: TopicPartition) => Offset(45).some).some,
4448
)
4549

4650
val result = wm.preCommit(Map(topicPartition -> new OffsetAndMetadata(999)))

‎kafka-connect-aws-s3/src/main/scala/io/lenses/streamreactor/connect/aws/s3/sink/config/S3SinkConfig.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ object S3SinkConfig extends PropsToConfigConverter[S3SinkConfig] {
7070
case class S3SinkConfig(
7171
connectionConfig: S3ConnectionConfig,
7272
bucketOptions: Seq[CloudSinkBucketOptions] = Seq.empty,
73-
indexOptions: IndexOptions,
73+
indexOptions: Option[IndexOptions],
7474
compressionCodec: CompressionCodec,
7575
batchDelete: Boolean,
7676
errorPolicy: ErrorPolicy,

‎kafka-connect-aws-s3/src/test/scala/io/lenses/streamreactor/connect/aws/s3/config/S3ConfigSettingsTest.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ class S3ConfigSettingsTest extends AnyFlatSpec with Matchers with LazyLogging {
3535
val configKeys =
3636
S3SinkConfigDef.config.configKeys().keySet().asScala ++ S3SourceConfigDef.config.configKeys().keySet().asScala
3737

38-
configKeys.size shouldBe 53
38+
configKeys.size shouldBe 54
3939
configKeys.foreach {
4040
k => k.toLowerCase should be(k)
4141
}

‎kafka-connect-azure-datalake/src/main/scala/io/lenses/streamreactor/connect/datalake/sink/config/DatalakeSinkConfig.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ object DatalakeSinkConfig extends PropsToConfigConverter[DatalakeSinkConfig] {
6565
case class DatalakeSinkConfig(
6666
connectionConfig: AzureConnectionConfig,
6767
bucketOptions: Seq[CloudSinkBucketOptions] = Seq.empty,
68-
indexOptions: IndexOptions,
68+
indexOptions: Option[IndexOptions],
6969
compressionCodec: CompressionCodec,
7070
errorPolicy: ErrorPolicy,
7171
connectorRetryConfig: RetryConfig,

‎kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/config/IndexSettings.scala‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,11 @@ trait IndexConfigKeys extends WithConnectorPrefix {
3434
s"Name of the indexes directory"
3535
private val INDEXES_DIRECTORY_NAME_DEFAULT = ".indexes"
3636

37+
val ENABLE_EXACTLY_ONCE = s"$connectorPrefix.exactly.once.enable"
38+
private val ENABLE_EXACTLY_ONCE_DOC =
39+
s"Exactly once is enabled by default. It works by keeping an .indexes directory at the root of your bucket with subdirectories for indexes. Exactly once support can be disabled and the default offset tracking from kafka can be used instead by setting this to false."
40+
private val ENABLE_EXACTLY_ONCE_DEFAULT = true
41+
3742
def addIndexSettingsToConfigDef(configDef: ConfigDef): ConfigDef =
3843
configDef
3944
.define(
@@ -58,11 +63,22 @@ trait IndexConfigKeys extends WithConnectorPrefix {
5863
ConfigDef.Width.LONG,
5964
INDEXES_DIRECTORY_NAME,
6065
)
66+
.define(
67+
ENABLE_EXACTLY_ONCE,
68+
Type.BOOLEAN,
69+
ENABLE_EXACTLY_ONCE_DEFAULT,
70+
Importance.LOW,
71+
ENABLE_EXACTLY_ONCE_DOC,
72+
"Sink Seek",
73+
3,
74+
ConfigDef.Width.NONE,
75+
ENABLE_EXACTLY_ONCE,
76+
)
6177
}
6278
trait IndexSettings extends BaseSettings with IndexConfigKeys {
63-
def getIndexSettings: IndexOptions =
64-
IndexOptions(
79+
def getIndexSettings: Option[IndexOptions] =
80+
Option.when(getBoolean(ENABLE_EXACTLY_ONCE))(IndexOptions(
6581
getInt(SEEK_MAX_INDEX_FILES),
6682
getString(INDEXES_DIRECTORY_NAME),
67-
)
83+
))
6884
}

‎kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/config/traits/CloudConfig.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ trait CloudSinkConfig[CC] extends CloudConfig {
5656
*
5757
* @return The offset seeker options for the cloud sink.
5858
*/
59-
def indexOptions: IndexOptions
59+
def indexOptions: Option[IndexOptions]
6060

6161
/**
6262
* Retrieves the compression codec for the cloud sink.

‎kafka-connect-cloud-common/src/main/scala/io/lenses/streamreactor/connect/cloud/common/sink/CloudSinkTask.scala‎

Lines changed: 42 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import io.lenses.streamreactor.connect.cloud.common.model.Topic
3232
import io.lenses.streamreactor.connect.cloud.common.model.TopicPartition
3333
import io.lenses.streamreactor.connect.cloud.common.sink.conversion.HeaderToStringConverter
3434
import io.lenses.streamreactor.connect.cloud.common.sink.conversion.ValueToSinkDataConverter
35+
import io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManager
3536
import io.lenses.streamreactor.connect.cloud.common.sink.writer.WriterManager
3637
import io.lenses.streamreactor.connect.cloud.common.storage.FileMetadata
3738
import io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface
@@ -69,8 +70,10 @@ abstract class CloudSinkTask[MD <: FileMetadata, C <: CloudSinkConfig[CC], CC <:
6970
private val writerManagerCreator = new WriterManagerCreator[MD, C]()
7071

7172
private var logMetrics = false
72-
private var writerManager: WriterManager[MD] = _
73-
implicit var connectorTaskId: ConnectorTaskId = _
73+
private var writerManager: WriterManager[MD] = _
74+
private var maybeIndexManager: Option[IndexManager[MD]] = _
75+
76+
implicit var connectorTaskId: ConnectorTaskId = _
7477

7578
override def version(): String = manifest.getVersion()
7679

@@ -89,7 +92,11 @@ abstract class CloudSinkTask[MD <: FileMetadata, C <: CloudSinkConfig[CC], CC <:
8992
val props = MapUtils.mergeProps(contextProps, fallbackProps.asScala.toMap)
9093
val errOrWriterMan = createWriterMan(props)
9194

92-
errOrWriterMan.leftMap(throw _).foreach(writerManager = _)
95+
errOrWriterMan.leftMap(throw _).foreach {
96+
case (mim, wm) =>
97+
maybeIndexManager = mim
98+
writerManager = wm
99+
}
93100
}
94101

95102
private def rollback(topicPartitions: Set[TopicPartition]): Unit =
@@ -217,29 +224,30 @@ abstract class CloudSinkTask[MD <: FileMetadata, C <: CloudSinkConfig[CC], CC <:
217224
actualOffsets
218225
}
219226

220-
override def open(partitions: util.Collection[KafkaTopicPartition]): Unit = {
221-
val partitionsDebug = partitions.asScala.map(tp => s"${tp.topic()}-${tp.partition()}").mkString(",")
222-
logger.debug(s"[{}] Open partitions", connectorTaskId.show, partitionsDebug: Any)
223-
224-
val topicPartitions = partitions.asScala
225-
.map(tp => TopicPartition(Topic(tp.topic), tp.partition))
226-
.toSet
227-
228-
handleErrors(
229-
for {
230-
tpoMap <- writerManager.open(topicPartitions)
231-
} yield {
232-
tpoMap.foreach {
233-
case (topicPartition, offset) =>
234-
logger.debug(
235-
s"[${connectorTaskId.show}] Seeking to ${topicPartition.topic.value}-${topicPartition.partition}:${offset.value}",
236-
)
237-
context.offset(topicPartition.toKafka, offset.value)
238-
}
239-
},
240-
)
241-
242-
}
227+
override def open(partitions: util.Collection[KafkaTopicPartition]): Unit =
228+
maybeIndexManager.foreach {
229+
indexManager =>
230+
val partitionsDebug = partitions.asScala.map(tp => s"${tp.topic()}-${tp.partition()}").mkString(",")
231+
logger.debug(s"[{}] Open partitions", connectorTaskId.show, partitionsDebug: Any)
232+
233+
val topicPartitions = partitions.asScala
234+
.map(tp => TopicPartition(Topic(tp.topic), tp.partition))
235+
.toSet
236+
237+
handleErrors(
238+
for {
239+
tpoMap <- indexManager.open(topicPartitions)
240+
} yield {
241+
tpoMap.foreach {
242+
case (topicPartition, offset) =>
243+
logger.debug(
244+
s"[${connectorTaskId.show}] Seeking to ${topicPartition.topic.value}-${topicPartition.partition}:${offset.value}",
245+
)
246+
context.offset(topicPartition.toKafka, offset.value)
247+
}
248+
},
249+
)
250+
}
243251

244252
/**
245253
* Whenever close is called, the topics and partitions assigned to this task
@@ -269,17 +277,21 @@ abstract class CloudSinkTask[MD <: FileMetadata, C <: CloudSinkConfig[CC], CC <:
269277

270278
def convertPropsToConfig(connectorTaskId: ConnectorTaskId, props: Map[String, String]): Either[Throwable, C]
271279

272-
private def createWriterMan(props: Map[String, String]): Either[Throwable, WriterManager[MD]] =
280+
private def createWriterMan(
281+
props: Map[String, String],
282+
): Either[Throwable, (Option[IndexManager[MD]], WriterManager[MD])] =
273283
for {
274284
config <- convertPropsToConfig(connectorTaskId, props)
275285
s3Client <- createClient(config.connectionConfig)
276286
storageInterface = createStorageInterface(connectorTaskId, config, s3Client)
277287
_ <- setRetryInterval(config)
278-
writerManager <- Try(writerManagerCreator.from(config)(connectorTaskId, storageInterface)).toEither
279-
_ <- initializeFromConfig(config)
288+
(maybeIndexManager, writerManager) <- Try(
289+
writerManagerCreator.from(config)(connectorTaskId, storageInterface),
290+
).toEither
291+
_ <- initializeFromConfig(config)
280292
} yield {
281293
logMetrics = config.logMetrics
282-
writerManager
294+
(maybeIndexManager, writerManager)
283295
}
284296

285297
private def initializeFromConfig(config: C): Either[Throwable, Unit] =

0 commit comments

Comments
 (0)