Use batch delete operation when cleaning up blob ids (#215)

* First pass at using the batch delete api

* Be a bit smarter about what we remove from deletableBlobIds

* Correct a few syntax errors

* Handle the case there are no blobs to delete

* Tweak target index for removing failed deletes

* Correct issues with long->int conversion

Author: farmdawgnation

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/GCSToBQLoadRunnable.java

@@ -263,19 +263,45 @@ private void checkJobs() {
    */
   private void deleteBlobs() {
     List<BlobId> blobIdsToDelete = new ArrayList<>();
-    logger.info("Attempting to delete {} blobs", deletableBlobIds.size());
-    for (BlobId blobId : deletableBlobIds) {
-      try {
-        bucket.getStorage().delete(blobId);
-        blobIdsToDelete.add(blobId);
-      } catch (StorageException ex) {
-        logger.warn("Failed to delete blob: {}/{}", blobId.getBucket(), blobId.getName());
+    blobIdsToDelete.addAll(deletableBlobIds);
+    int numberOfBlobs = blobIdsToDelete.size();
+    int failedDeletes = 0;
+    int successfulDeletes = 0;
+
+    if (numberOfBlobs == 0) {
+      logger.info("No blobs to delete");
+      return;
+    }
+
+    logger.info("Attempting to delete {} blobs", numberOfBlobs);
+
+    try {
+      // Issue a batch delete api call
+      List<Boolean> resultList = bucket.getStorage().delete(blobIdsToDelete);
+
+      // Filter the blobs we couldn't delete from the list of deletable blobs
+      for (int i = 0; i < numberOfBlobs; i++) {
+        if (!resultList.get(i)) {
+          // This blob was not successful, remove it from the list.
+          // Adjust the target index by the number of failed deletes we've
+          // already seen since we're mutating the list as we go.
+          int targetIndex = i - failedDeletes;
+          blobIdsToDelete.remove(targetIndex);
+          failedDeletes++;
+        }
       }
+
+      // Calculate number of successful deletes, remove the successful deletes from
+      // the deletableBlobIds.
+      successfulDeletes = numberOfBlobs - failedDeletes;
+      deletableBlobIds.removeAll(blobIdsToDelete);
+
+      logger.info("Successfully deleted {} blobs; failed to delete {} blobs",
+                  successfulDeletes,
+                  failedDeletes);
+    } catch (StorageException ex) {
+      logger.warn("Storage exception while attempting to delete blobs", ex);
     }
-    deletableBlobIds.removeAll(blobIdsToDelete);
-    logger.info("Successfully deleted {} blobs; failed to delete {} blobs",
-                blobIdsToDelete.size(),
-                deletableBlobIds.size());
   }
 
   @Override