-
Notifications
You must be signed in to change notification settings - Fork 369
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Retain directories after post-processing (#197)
* Retain directories after post-processing Customers have noticed that when the Lenses S3 Source Connector deletes files on S3, then the entire path is deleted. Whilst this is actually due to the way S3 works, we can actually do something about this in the connector. This adds a new boolean KCQL property to the S3 source: `post.process.action.retain.dirs` (default value: `false`) If this is set to `true`, then upon moving/deleting files within the source post processing, then first a zero-byte object will be created to ensure that the path will still be represented on S3.
- Loading branch information
1 parent
c3920a3
commit 7e6e823
Showing
13 changed files
with
240 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
53 changes: 53 additions & 0 deletions
53
...ain/scala/io/lenses/streamreactor/connect/cloud/common/source/config/DirectoryCache.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| /* | ||
| * Copyright 2017-2025 Lenses.io Ltd | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.lenses.streamreactor.connect.cloud.common.source.config | ||
|
|
||
| import cats.implicits.catsSyntaxEitherId | ||
| import io.lenses.streamreactor.connect.cloud.common.storage.FileCreateError | ||
| import io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface | ||
|
|
||
| import scala.collection.mutable | ||
|
|
||
| /** | ||
| * A cache for tracking directories that have been created. | ||
| * | ||
| * @param storageInterface The storage interface used to create directories. | ||
| */ | ||
| class DirectoryCache(storageInterface: StorageInterface[_]) { | ||
|
|
||
| // A mutable set to keep track of created directories. | ||
| private val directoriesCreated = mutable.Set[(String, String)]() | ||
|
|
||
| /** | ||
| * Ensures that a directory exists in the specified bucket and path. | ||
| * | ||
| * @param bucket The bucket in which the directory should exist. | ||
| * @param path The path of the directory to check or create. | ||
| * @return Either a FileCreateError if the directory creation failed, or Unit if the directory exists or was created successfully. | ||
| */ | ||
| def ensureDirectoryExists(bucket: String, path: String): Either[FileCreateError, Unit] = | ||
| if (directoriesCreated.contains((bucket, path))) { | ||
| ().asRight | ||
| } else { | ||
| storageInterface.createDirectoryIfNotExists(bucket, path) match { | ||
| case Left(value) => value.asLeft | ||
| case Right(_) => | ||
| directoriesCreated.add((bucket, path)) | ||
| ().asRight | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
57 changes: 57 additions & 0 deletions
57
...scala/io/lenses/streamreactor/connect/cloud/common/source/config/DirectoryCacheTest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| /* | ||
| * Copyright 2017-2025 Lenses.io Ltd | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.lenses.streamreactor.connect.cloud.common.source.config | ||
|
|
||
| import io.lenses.streamreactor.connect.cloud.common.storage.FileCreateError | ||
| import io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface | ||
| import org.mockito.MockitoSugar | ||
| import org.scalatest.funsuite.AnyFunSuiteLike | ||
| import org.scalatest.matchers.should.Matchers | ||
|
|
||
| class DirectoryCacheTest extends AnyFunSuiteLike with Matchers with MockitoSugar { | ||
|
|
||
| test("ensureDirectoryExists should return Right(Unit) and add the directory to the cache if it does not exist") { | ||
| val storageInterface = mock[StorageInterface[_]] | ||
| when(storageInterface.createDirectoryIfNotExists("bucket", "path")).thenReturn(Right(())) | ||
| val cache = new DirectoryCache(storageInterface) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Right(())) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Right(())) | ||
| verify(storageInterface, times(1)).createDirectoryIfNotExists("bucket", "path") | ||
| } | ||
|
|
||
| test( | ||
| "ensureDirectoryExists should return Left(FileCreateError) and not add to the cache if creating the directory fails", | ||
| ) { | ||
| val storageInterface = mock[StorageInterface[_]] | ||
| val error = FileCreateError(new IllegalStateException("Bad"), "data") | ||
| when(storageInterface.createDirectoryIfNotExists("bucket", "path")).thenReturn(Left(error)) | ||
| val cache = new DirectoryCache(storageInterface) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Left(error)) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Left(error)) | ||
| verify(storageInterface, times(2)).createDirectoryIfNotExists("bucket", "path") | ||
| } | ||
|
|
||
| test("ensureDirectoryExists should not add the directory to the cache if it already exists") { | ||
| val storageInterface = mock[StorageInterface[_]] | ||
| when(storageInterface.createDirectoryIfNotExists("bucket", "path")).thenReturn(Right(())) | ||
| val cache = new DirectoryCache(storageInterface) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Right(())) | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Right(())) | ||
| verify(storageInterface, times(1)).createDirectoryIfNotExists("bucket", "path") | ||
| cache.ensureDirectoryExists("bucket", "path") should be(Right(())) | ||
| verify(storageInterface, times(1)).createDirectoryIfNotExists("bucket", "path") | ||
| } | ||
| } |
Oops, something went wrong.