Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Only list gcs folder blobs in GCSToBQLoadRunnable #270

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Only list gcs folder blobs in GCSToBQLoadRunnable #270

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
015c272
Only list gcs folder blobs in GCSToBQLoadRunnable
May 12, 2020
90e63b6
Test gcs folder isolation
Jun 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQuerySinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,7 @@ private void startGCSToBQLoadTask() {
logger.info("Attempting to start GCS Load Executor.");
gcsLoadExecutor = Executors.newScheduledThreadPool(1);
String bucketName = config.getString(config.GCS_BUCKET_NAME_CONFIG);
String folderName = config.getString(config.GCS_FOLDER_NAME_CONFIG);
Storage gcs = getGcs();
// get the bucket, or create it if it does not exist.
Bucket bucket = gcs.get(bucketName);
Expand All @@ -359,7 +360,8 @@ private void startGCSToBQLoadTask() {
BucketInfo bucketInfo = BucketInfo.of(bucketName);
bucket = gcs.create(bucketInfo);
}
GCSToBQLoadRunnable loadRunnable = new GCSToBQLoadRunnable(getBigQuery(), bucket);
Storage.BlobListOption folderPrefixOption = Storage.BlobListOption.prefix(folderName);
GCSToBQLoadRunnable loadRunnable = new GCSToBQLoadRunnable(getBigQuery(), bucket, folderPrefixOption);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does this work as expected if there is no folder prefix configured?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe so. As the GCS_FOLDER_NAME_DEFAULT on the config is set to an empty string, we are filtering for objects with a path prefixed by an empty string which is the same as no filter at all.

At least that was how I thought it would work, and took the passing integration tests as confirmation :)


int intervalSec = config.getInt(BigQuerySinkConfig.BATCH_LOAD_INTERVAL_SEC_CONFIG);
gcsLoadExecutor.scheduleAtFixedRate(loadRunnable, intervalSec, intervalSec, TimeUnit.SECONDS);
Expand Down
7 changes: 5 additions & 2 deletions kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/GCSToBQLoadRunnable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.StorageException;
import com.google.cloud.storage.Storage.BlobListOption;

import com.wepay.kafka.connect.bigquery.write.row.GCSToBQWriter;

Expand Down Expand Up @@ -58,6 +59,7 @@ public class GCSToBQLoadRunnable implements Runnable {

private final BigQuery bigQuery;
private final Bucket bucket;
private final BlobListOption folderPrefixOption;
private final Map<Job, List<BlobId>> activeJobs;
private final Set<BlobId> claimedBlobIds;
private final Set<BlobId> deletableBlobIds;
Expand All @@ -79,9 +81,10 @@ public class GCSToBQLoadRunnable implements Runnable {
* @param bigQuery the {@link BigQuery} instance.
* @param bucket the the GCS bucket to read from.
*/
public GCSToBQLoadRunnable(BigQuery bigQuery, Bucket bucket) {
public GCSToBQLoadRunnable(BigQuery bigQuery, Bucket bucket, BlobListOption folderPrefixOption) {
this.bigQuery = bigQuery;
this.bucket = bucket;
this.folderPrefixOption = folderPrefixOption;
this.activeJobs = new HashMap<>();
this.claimedBlobIds = new HashSet<>();
this.deletableBlobIds = new HashSet<>();
Expand All @@ -101,7 +104,7 @@ private Map<TableId, List<Blob>> getBlobsUpToLimit() {
Map<TableId, Long> tableToCurrentLoadSize = new HashMap<>();

logger.trace("Starting GCS bucket list");
Page<Blob> list = bucket.list();
Page<Blob> list = bucket.list(folderPrefixOption);
logger.trace("Finished GCS bucket list");

for (Blob blob : list.iterateAll()) {
Expand Down