[GitHub] [druid] TSFenwick opened a new pull request, #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick opened a new pull request, #14131:
URL: https://github.com/apache/druid/pull/14131

   … one by one
   
   create new batchdelete method in datasegment killer that has default functionality of iterating through all segments and calling delete on them. This will enable a slow rollout of other deepstorage implementations to move to a batched delete on their own time
   
   <!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
   
   <!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
   
   Fixes #XXXX.
   
   <!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
   
   <!-- If you are a committer, follow the PR action item checklist for committers:
   https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
   
   ### Description
   
   <!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
   
   <!-- Describe your patch: what did you change in code? How did you fix the problem? -->
   
   <!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
   
   #### Fixed the bug ...
   #### Renamed the class ...
   #### Added a forbidden-apis entry ...
   
   <!--
   In each section, please describe design decisions made, including:
    - Choice of algorithms
    - Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
    - Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
    - Method organization and design (how the logic is split between methods, parameters and return types)
    - Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
   -->
   
   
   <!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
   
   <!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
   
   <!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
   
   #### Release note
   <!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users. 
   
   If your change doesn't have end user impact, you can skip this section.
   
   For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
   
   -->
   
   
   <hr>
   
   ##### Key changed/added classes in this PR
    * `MyFoo`
    * `OurBar`
    * `TheirBaz`
   
   <hr>
   
   <!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
   
   This PR has:
   
   - [ ] been self-reviewed.
      - [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
   - [ ] added documentation for new or modified features or behaviors.
   - [ ] a release note entry in the PR description.
   - [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
   - [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
   - [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
   - [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
   - [ ] added integration tests.
   - [ ] been tested in a test Druid cluster.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1178622255


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {
+        throw new SegmentLoadingException(e, "Couldn't kill all segment but deleted[%s]: [%s]", e.getDeletedObjects(), e);

Review Comment:
   i was following the pattern set forth by how segments were deleted before this change. I can change it to do all, and then throw an exception that contains all exceptions that were thrown.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {

Review Comment:
   ok



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182824999


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,81 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    // don't fail immediately delete as many as possible then we fail in the end by throwing one exception that has all
+    // exceptions in it
+    List<Exception> exceptions = new ArrayList<>();
+    // s3 jdk is weird where the call to delete is a 200 status code for the response in the api, but the call is unable
+    // to delete some segments is an exception. this exception has a list of all objects it couldnt' delete. If you make
+    // a call that is a 401 or 403 outright the exception is thrown, but there is no list of objects it couldnt' delete.
+    // so we store them here to log later.
+    List<String> unableToDeleteKeysButNotStoredInException = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: %s the following index files: %s from s3!", s3Bucket, keysToDeleteStrings);

Review Comment:
   Removing from bucket: [%s] the following index files: [%s] from s3!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] rohangarg commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["rohangarg (via GitHub)" <gi...@apache.org>]

rohangarg commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1239381102


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e)
+      {
+        keysNotDeleted.addAll(e.getErrors().stream()
+         .map(MultiObjectDeleteException.DeleteError::getKey)
+         .collect(Collectors.toList()));
+      }
+      catch (AmazonServiceException e)
+      {
+        throw new SegmentLoadingException(e,

Review Comment:
   Cool, thanks for the explanation @TSFenwick, this makes sense to me



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1271750433


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();

Review Comment:
   Those shouldn't be there I'll remove them 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] gianm commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["gianm (via GitHub)" <gi...@apache.org>]

gianm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1180574899


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   As to testing: one thing to think about is whether this uses any new S3 APIs and therefore may require new IAM permissions. If it would require new IAM permissions then we definitely need a feature flag and we need to call it out in the docs and release notes. Or, better, adjust the design to not use new APIs and not need new IAM permissions.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1210847860


##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments does not exist in deep storage, the method should not throw an exception.
+   * <p>
+   * This version of kill must **NOT** require additional permissions on the deep storage beyond what
+   * {@link #kill(DataSegment)} requires. requires.

Review Comment:
   typo. remove extra requires.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] suneet-s commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["suneet-s (via GitHub)" <gi...@apache.org>]

suneet-s commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1181830073


##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments do not exist in deep storage method should not throw an exception.
+   * <p>
+   * This version Kill must not require additional permissions on the deep storage beyond what the single-arg
+   * form of kill @link org.apache.druid.segment.loading.DataSegmentKiller requires.

Review Comment:
   ```suggestion
      * This version of kill must **NOT** require additional permissions on the deep storage beyond what 
      * {@link #kill(DataSegment)} requires.
   ```



##########
server/src/test/java/org/apache/druid/segment/loading/OmniDataSegmentKillerTest.java:
##########
@@ -119,6 +139,31 @@ public void testKillTombstone() throws Exception
     segmentKiller.kill(tombstone);
   }
 
+  @Test
+  public void testKillMultipleSegmentsWithType() throws SegmentLoadingException
+  {
+    final DataSegmentKiller killerSane = Mockito.mock(DataSegmentKiller.class);
+    final DataSegmentKiller killerSaneTwo = Mockito.mock(DataSegmentKiller.class);
+    final DataSegment segment1 = Mockito.mock(DataSegment.class);
+    final DataSegment segment2 = Mockito.mock(DataSegment.class);
+    final DataSegment segment3 = Mockito.mock(DataSegment.class);
+    Mockito.when(segment1.isTombstone()).thenReturn(false);
+    Mockito.when(segment1.getLoadSpec()).thenReturn(ImmutableMap.of("type", "sane"));
+    Mockito.when(segment2.isTombstone()).thenReturn(false);
+    Mockito.when(segment2.getLoadSpec()).thenReturn(ImmutableMap.of("type", "sane"));
+    Mockito.when(segment3.isTombstone()).thenReturn(false);
+    Mockito.when(segment3.getLoadSpec()).thenReturn(ImmutableMap.of("type", "sane_2"));
+
+    final Injector injector = createInjectorFromMap(ImmutableMap.of("sane", killerSane, "sane_2", killerSaneTwo));
+    final OmniDataSegmentKiller segmentKiller = injector.getInstance(OmniDataSegmentKiller.class);
+    segmentKiller.kill(ImmutableList.of(segment1, segment2, segment3));
+
+    Mockito.verify(killerSane, Mockito.times(1))
+           .kill((List<DataSegment>) argThat(containsInAnyOrder(segment1, segment2)));
+    Mockito.verify(killerSaneTwo, Mockito.times(1))
+           .kill((List<DataSegment>) argThat(containsInAnyOrder(segment3)));
+  }
+

Review Comment:
   Add tests for error handling please



##########
extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentKillerTest.java:
##########
@@ -213,4 +219,96 @@ public void test_killAll_nonrecoverableExceptionWhenListingObjects_deletesAllSeg
     Assert.assertTrue(ioExceptionThrown);
     EasyMock.verify(s3Client, segmentPusherConfig, inputDataConfig);
   }
+
+  @Test
+  public void test_kill_singleSegment_doesntexist_passes() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(false);
+    EasyMock.expectLastCall().once();
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(false);
+    EasyMock.expectLastCall().once();
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(DATA_SEGMENT);
+  }
+
+  @Test
+  public void test_kill_singleSegment_exists_passes() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(DATA_SEGMENT);
+  }
+
+  @Test
+  public void test_kill_listOfOneSegment() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of(DATA_SEGMENT));
+  }
+
+  @Test
+  public void test_kill_listOfNoSegments() throws SegmentLoadingException
+  {
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of());
+    // has an assertion error if there is an unexpected method call on a mock. Do nothing because we expect the kill
+    // method to not interact with mocks
+  }
+
+  @Test
+  public void test_kill_listOfSegments() throws SegmentLoadingException
+  {
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(TEST_BUCKET);
+    deleteObjectsRequest.withKeys(KEY_1_PATH, KEY_1_PATH);
+    // struggled with the idea of making it match on equaling this
+    s3Client.deleteObjects(EasyMock.anyObject(DeleteObjectsRequest.class));
+    EasyMock.expectLastCall().andVoid().times(2);
+
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of(DATA_SEGMENT, DATA_SEGMENT));
+  }
+
+  private static final DataSegment DATA_SEGMENT = new DataSegment(

Review Comment:
   Constants should be defined at the top of the file



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["jasonk000 (via GitHub)" <gi...@apache.org>]

jasonk000 commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1644914425

   @TSFenwick we'd spotted the same in #14634 , please @ me if you'd like any help to close this out, extra testing, etc, as this would save me a lot of effort too (~100 million segments pending to delete).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172741352


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            segmentsChunk.stream()
+                         .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                         .toArray(String[]::new));
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            segmentsChunk.stream()
+                         .map(segment -> DataSegmentKiller.descriptorPath(MapUtils.getString(
+                             segment.getLoadSpec(),
+                             "key"
+                         )))
+                         .toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+    }
+    catch (MultiObjectDeleteException e) {

Review Comment:
   what other exceptions can happen here? Is this the best message in the exception?
   
   https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3.html#deleteObjects-com.amazonaws.services.s3.model.DeleteObjectsRequest-
   https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] suneet-s commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["suneet-s (via GitHub)" <gi...@apache.org>]

suneet-s commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1178452319


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());

Review Comment:
   ```suggestion
           log.info("Removing indexes from s3, bucket: %s, keys [%s]", s3Bucket, keys);
   ```
   
   Can you paste an example of what this line looks like?



##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,19 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Removes a list of segment files (indexes and metadata) from deep storage. This method can be more efficient if
+   * implementer of this interface leverages batch/bulk deletes.

Review Comment:
   nit:
   ```suggestion
         Removes a list of segment files (indexes and metadata) from deep storage. This method can be more efficient if
      * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
      * Implementers of this interface can leverage batch / bulk deletes to be more efficient.
   ```



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {

Review Comment:
   Instead of making 2 `DeleteObjectsRequest`s can we combine all the keys into one list of keys and then split the list based on the max 1000 keys that can be deleted at a time.
   
   This will also mean that descriptor files will not be left behind if deleting one segment fails.
   
   At this point the segments are already removed from the metadata store so the files and their descriptors should no longer be referenced anywhere in Driud.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);

Review Comment:
   We don't care about the successes since we already have the list of keys in the request.
   
   ```suggestion
           deleteObjectsRequest = deleteObjectsRequest.withKeys(keys).withQuiet(true);
   ```



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);

Review Comment:
   ```suggestion
   ```



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);

Review Comment:
   ```suggestion
   ```



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   If we keep this feature flag, it should be enabled by default as this is the better method. Since there is no change in behavior with the flag enabled vs disabled, I don't see a reason to default disable this behavior when batch deleting segments is the more optimal way to delete segments. We should also document this in the task context documentation and mention that this feature flag will be going away in the very near future.
   
   Is there a way to configure the auto-kill to specify this context so that Druid operators who have auto-kill enabled can enable / disable this behavior for kill tasks generated from the auto-kill functionality in Druid?



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {
+        throw new SegmentLoadingException(e, "Couldn't kill all segment but deleted[%s]: [%s]", e.getDeletedObjects(), e);

Review Comment:
   I don't think we should throw an exception if deleting some of the segments fail. We should run through all the objects to delete and then at the end, throw an exception with the list of objects that failed to delete.
   
   ```suggestion
           throw new SegmentLoadingException(e, "Failed to delete keys [%s]", e.getDeletedObjects().stream().map(DeleteObjectsResult.DeletedObject::getKey).collect(Collectors.toList())));
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["maytasm (via GitHub)" <gi...@apache.org>]

maytasm merged PR #14131:
URL: https://github.com/apache/druid/pull/14131


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] suneet-s commented on pull request #14131: Speed up kill tasks by deleting segments in batch

["suneet-s (via GitHub)" <gi...@apache.org>]

suneet-s commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1591741045

   @gianm / @rohangarg any other comments on this patch?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1210847538


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   @gianm I decided to enable this by default and not have a feature flag for it at all. The api that is used to delete multiple keys from a bucket is already used elsewhere in the extension. 
   
   I added that verbiage for kill not require additional permissions



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1263042110


##########
extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentKillerTest.java:
##########
@@ -347,7 +347,7 @@ public void test_kill_listOfSegments_multiDeleteExceptionIsThrown()
       segmentKiller.kill(ImmutableList.of(DATA_SEGMENT_1, DATA_SEGMENT_2));
     }
     catch (SegmentLoadingException exc) {
-      Assert.assertEquals("Couldn't delete from bucket: [test_bucket] these files [[key1/]]", exc.getMessage());
+      Assert.assertEquals("Couldn't delete segments from s3 see logs for more details", exc.getMessage());

Review Comment:
   will do



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172747274


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {

Review Comment:
   add log messages to more closely match the kill task logging



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1190186692


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);

Review Comment:
   i can do that. will want to keep the retry count low, as a successful delete call can throw an exception if a single segment fails to delete for a "valid" reason in aws's eyes



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["maytasm (via GitHub)" <gi...@apache.org>]

maytasm commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1648771151

   This PR now has three +1 from three committers. 
   
   @TSFenwick Can you check the failing tests?
   
   @rohangarg @gianm Do you have any other comments or feel that this PR should not be merged? 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] gianm commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["gianm (via GitHub)" <gi...@apache.org>]

gianm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1180574086


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   IMO enabling this new feature by default is best, as it's better and the risk isn't super high.
   
   I'm not even sure we need a feature flag at all. We should be able to get the risk low enough with appropriate testing that a flag isn't needed.
   
   If we're gonna have one, then yeah, it needs to be settable for people that use autokill.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182842581


##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments do not exist in deep storage method should not throw an exception.

Review Comment:
   "does not exist"



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] gianm commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["gianm (via GitHub)" <gi...@apache.org>]

gianm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1180577040


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   If we decide to go the "not require new APIs" route — which IMO is best — then we should have a note in the interface that says that the multi-arg form of `kill` must not require additional permissions on the deep storage beyond what the single-arg form of `kill` requires. Otherwise, when a multi-arg impl is contributed for another deep storage, it could break people in production that don't have the right set of permissions.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182839947


##########
extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentKillerTest.java:
##########
@@ -213,4 +219,96 @@ public void test_killAll_nonrecoverableExceptionWhenListingObjects_deletesAllSeg
     Assert.assertTrue(ioExceptionThrown);
     EasyMock.verify(s3Client, segmentPusherConfig, inputDataConfig);
   }
+
+  @Test
+  public void test_kill_singleSegment_doesntexist_passes() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(false);
+    EasyMock.expectLastCall().once();
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(false);
+    EasyMock.expectLastCall().once();
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(DATA_SEGMENT);
+  }
+
+  @Test
+  public void test_kill_singleSegment_exists_passes() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(DATA_SEGMENT);
+  }
+
+  @Test
+  public void test_kill_listOfOneSegment() throws SegmentLoadingException
+  {
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+    EasyMock.expect(s3Client.doesObjectExist(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH)).andReturn(true);
+    EasyMock.expectLastCall().once();
+
+    s3Client.deleteObject(TEST_BUCKET, KEY_1_DESCRIPTOR_PATH);
+    EasyMock.expectLastCall().andVoid();
+
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of(DATA_SEGMENT));
+  }
+
+  @Test
+  public void test_kill_listOfNoSegments() throws SegmentLoadingException
+  {
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of());
+    // has an assertion error if there is an unexpected method call on a mock. Do nothing because we expect the kill
+    // method to not interact with mocks
+  }
+
+  @Test
+  public void test_kill_listOfSegments() throws SegmentLoadingException
+  {
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(TEST_BUCKET);
+    deleteObjectsRequest.withKeys(KEY_1_PATH, KEY_1_PATH);
+    // struggled with the idea of making it match on equaling this
+    s3Client.deleteObjects(EasyMock.anyObject(DeleteObjectsRequest.class));
+    EasyMock.expectLastCall().andVoid().times(2);
+
+
+    EasyMock.replay(s3Client, segmentPusherConfig, inputDataConfig);
+    segmentKiller = new S3DataSegmentKiller(Suppliers.ofInstance(s3Client), segmentPusherConfig, inputDataConfig);
+    segmentKiller.kill(ImmutableList.of(DATA_SEGMENT, DATA_SEGMENT));
+  }
+

Review Comment:
   add negative test cases with MultiPartObject Delete Exception thrown, and AmazonServiceException Thrown



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1193209315


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e)
+      {
+        keysNotDeleted.addAll(e.getErrors().stream()
+         .map(MultiObjectDeleteException.DeleteError::getKey)
+         .collect(Collectors.toList()));
+      }
+      catch (AmazonServiceException e)
+      {
+        throw new SegmentLoadingException(e,

Review Comment:
   Thats not how its done in the single kill when i read it.
    single kill
   ```java
   throw new SegmentLoadingException(e, "Couldn't kill segment[%s]: [%s]", segment.getId(), e);
   ```
   
   What im trying to do here is balance this fine line of making the code understandable yet also give a reasonable log message. and also to avoid making unnecessary calls for when there are lots of segments to be deleted. The multiobject delete also acts differently than the single kill in that the exceptions are used slightly differently.
   
   a `MultiObjectDeleteException` is a successful delete call where an object that was requested to be deleted couldn't be deleted for a "valid" reason. for example if there is a restricted permission on one to all the objects. but the causes can be numerous. this exception also has a list of all the objects it couldn't delete.
   
   an AmazonServiceException is the kind of exception you get from a 400-500 status code. This doesn't give a list of all the objects it couldn't delete. if you get a 401/403 exception here there is no point to make lots of delete calls.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172745088


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);

Review Comment:
   does it make sense to check if object exists before calling delete on them? DeleteObjects wont complain if you call delete on an object that doesn't exist and will happily say it everything succeeded.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["maytasm (via GitHub)" <gi...@apache.org>]

maytasm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1271741533


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();

Review Comment:
   What does line 157-158 do?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272665659


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
+      }
+      catch (AmazonServiceException e) {
+        hadException = true;
+        log.noStackTrace().warn(e,

Review Comment:
   https://github.com/apache/druid/pull/14131#discussion_r1263027023



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272777356


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);

Review Comment:
   good find



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272641348


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
+      }
+      catch (AmazonServiceException e) {
+        hadException = true;
+        log.noStackTrace().warn(e,

Review Comment:
   I had it as an error before, but i understand why @gianm said it should be a warn, since we are continuing on without stopping the process



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182837137


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,81 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    // don't fail immediately delete as many as possible then we fail in the end by throwing one exception that has all
+    // exceptions in it
+    List<Exception> exceptions = new ArrayList<>();
+    // s3 jdk is weird where the call to delete is a 200 status code for the response in the api, but the call is unable
+    // to delete some segments is an exception. this exception has a list of all objects it couldnt' delete. If you make
+    // a call that is a 401 or 403 outright the exception is thrown, but there is no list of objects it couldnt' delete.
+    // so we store them here to log later.
+    List<String> unableToDeleteKeysButNotStoredInException = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: %s the following index files: %s from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (AmazonServiceException e) {

Review Comment:
   catch MultiObjectDeleteException, and build list of unable to delete and continue.
   
   catch AmazonServiceException
   throw segment loading exception immediately



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172742363


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            segmentsChunk.stream()
+                         .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))

Review Comment:
   does it make sense to turn key into a constant for the s3 code



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172736165


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(

Review Comment:
   dont process data into keys 2x. process into keys. and then leave data around to process into descriptor paths



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1588468002

   # Incorrect Access Key
   
   ```{
     "id": "api-issued_kill_wikipedia_3_gokfnfge_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T03:20:19.252Z",
     "groupId": "api-issued_kill_wikipedia_3_gokfnfge_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T03:20:19.252Z",
     "type": "kill",
     "createdTime": "2023-06-13T03:20:19.303Z",
     "queueInsertionTime": "1970-01-01T00:00:00.000Z",
     "statusCode": "FAILED",
     "status": "FAILED",
     "runnerStatusCode": "WAITING",
     "duration": 6799,
     "location": {
       "host": "localhost",
       "port": 8100,
       "tlsPort": -1
     },
     "dataSource": "wikipedia_3",
     "errorMsg": "org.apache.druid.segment.loading.SegmentLoadingException: Unable to delete from bucket [tomfenwick-t..."
   }
   ```
   
   [403_error_delete.txt](https://github.com/apache/druid/files/11730294/403_error_delete.txt)
   
   # Successful no errors
   
   ```
   {
     "id": "api-issued_kill_testPerms1_lkdbjnmf_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T02:51:14.744Z",
     "groupId": "api-issued_kill_testPerms1_lkdbjnmf_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T02:51:14.744Z",
     "type": "kill",
     "createdTime": "2023-06-13T02:51:14.756Z",
     "queueInsertionTime": "1970-01-01T00:00:00.000Z",
     "statusCode": "SUCCESS",
     "status": "SUCCESS",
     "runnerStatusCode": "WAITING",
     "duration": 8795,
     "location": {
       "host": "localhost",
       "port": 8100,
       "tlsPort": -1
     },
     "dataSource": "testPerms1",
     "errorMsg": null
   }
   ```
   
   [succesful_batch_delete.txt](https://github.com/apache/druid/files/11730331/succesful_batch_delete.txt)
   
   # unable to delete one index
   
   ```
   {
     "id": "api-issued_kill_wikipedia_2_niigolng_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T03:13:35.830Z",
     "groupId": "api-issued_kill_wikipedia_2_niigolng_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T03:13:35.830Z",
     "type": "kill",
     "createdTime": "2023-06-13T03:13:35.835Z",
     "queueInsertionTime": "1970-01-01T00:00:00.000Z",
     "statusCode": "FAILED",
     "status": "FAILED",
     "runnerStatusCode": "WAITING",
     "duration": 6810,
     "location": {
       "host": "localhost",
       "port": 8100,
       "tlsPort": -1
     },
     "dataSource": "wikipedia_2",
     "errorMsg": "org.apache.druid.segment.loading.SegmentLoadingException: Couldn't delete from bucket: [tomfenwick-t..."
   }
   ```
   [no_perms_on_single_file_to_be_deleted_log.txt](https://github.com/apache/druid/files/11730316/no_perms_on_single_file_to_be_deleted_log.txt)
   
   # Unable to delete all indices
   
   ```
   {
     "id": "api-issued_kill_wikipedia_1_nbnhlfdk_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T02:56:52.826Z",
     "groupId": "api-issued_kill_wikipedia_1_nbnhlfdk_1000-01-01T00:00:00.000Z_2023-06-12T00:00:00.000Z_2023-06-13T02:56:52.826Z",
     "type": "kill",
     "createdTime": "2023-06-13T02:56:52.829Z",
     "queueInsertionTime": "1970-01-01T00:00:00.000Z",
     "statusCode": "FAILED",
     "status": "FAILED",
     "runnerStatusCode": "WAITING",
     "duration": 8239,
     "location": {
       "host": "localhost",
       "port": 8100,
       "tlsPort": -1
     },
     "dataSource": "wikipedia_1",
     "errorMsg": "org.apache.druid.segment.loading.SegmentLoadingException: Couldn't delete from bucket: [tomfenwick-t..."
   }
   ```
   
   [no_perms_on_all_files_to_be_deleted_log.txt](https://github.com/apache/druid/files/11730324/no_perms_on_all_files_to_be_deleted_log.txt)
   
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] rohangarg commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["rohangarg (via GitHub)" <gi...@apache.org>]

rohangarg commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1239380397


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);

Review Comment:
   @TSFenwick : Yes, that makes sense to me



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] gianm commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["gianm (via GitHub)" <gi...@apache.org>]

gianm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1241502240


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,66 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.

Review Comment:
   I don't see anything that guarantees this. It's possible for segments for the same datasource to be in different buckets: for example, if the deep storage location is changed at some point, old segments stay where they are.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,66 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);

Review Comment:
   nit: 1000 should be a `static final`.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,66 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {

Review Comment:
   nit: code is a bit awkward; either reuse `size`, or replace the first `size == 0` with `segments.isEmpty()` and don't save `size` at all.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -31,14 +34,20 @@
 import org.apache.druid.timeline.DataSegment;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  *
  */
 public class S3DataSegmentKiller implements DataSegmentKiller
 {
   private static final Logger log = new Logger(S3DataSegmentKiller.class);
+  private static final String BUCKET = "bucket";

Review Comment:
   nit: better to use the one from `S3DataSegmentPuller`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1179296416


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);

Review Comment:
   ok will get rid of the return result on delete objects as it's not needed with this flag



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1263044256


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -76,63 +77,114 @@ public S3DataSegmentKiller(
   @Override
   public void kill(List<DataSegment> segments) throws SegmentLoadingException
   {
-    int size = segments.size();
-    if (size == 0) {
+    if (segments.isEmpty()) {
       return;
     }
     if (segments.size() == 1) {
       kill(segments.get(0));
       return;
     }
 
-    // we can assume that all segments are in the same bucket.
-    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
-    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
 
-    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
-            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
-            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
-                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
-            .collect(Collectors.toList());
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from s3 see logs for more details"

Review Comment:
   makes sense, will do.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["jasonk000 (via GitHub)" <gi...@apache.org>]

jasonk000 commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272591391


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);

Review Comment:
   We can avoid the extra temporary ArrayList like this: 
   ```
   List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>());
   keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
   keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
   ```



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
+      }
+      catch (AmazonServiceException e) {
+        hadException = true;
+        log.noStackTrace().warn(e,
+            "Unable to delete from bucket [%s], the following keys [%s]",
+            s3Bucket,
+            chunkOfKeys.stream().map(DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.joining(", "))
+        );
+      }
+    }
+    return hadException;
+  }
+
   @Override
   public void kill(DataSegment segment) throws SegmentLoadingException

Review Comment:
   Is there any reason to keep this specifically?, I much prefer your bulk implementation.
   
   We could remove it completely:
   ```
   public void kill(DataSegment segment) throws SegmentLoadingException
   {
     kill(Collections.singletonList(segment));
   }
   ```
   
   Alternatively, we can avoid the call to `doesObjectExist` and issue the delete directly, since same outcome is achieved.



##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments does not exist in deep storage, the method should not throw an exception.
+   * <p>
+   * This version of kill must **NOT** require additional permissions on the deep storage beyond what
+   * {@link #kill(DataSegment)} requires.
+   * @param segments The list of segments to kill.
+   * @throws SegmentLoadingException If there is an exception during deletion such as a segment in the list could not be
+   * completely removed.
+   */
+  default void kill(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   for example [here](https://github.com/apache/druid/blob/28914bbab82480f33a345913a30c32321d36fc37/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BaseAppenderatorDriver.java#L505)



##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments does not exist in deep storage, the method should not throw an exception.
+   * <p>
+   * This version of kill must **NOT** require additional permissions on the deep storage beyond what
+   * {@link #kill(DataSegment)} requires.
+   * @param segments The list of segments to kill.
+   * @throws SegmentLoadingException If there is an exception during deletion such as a segment in the list could not be
+   * completely removed.
+   */
+  default void kill(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   We could consider extending this to `killQuietly`, as that is called from the appenderator layer for multiple segments at a time.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
+      }
+      catch (AmazonServiceException e) {
+        hadException = true;
+        log.noStackTrace().warn(e,

Review Comment:
   IMO this should be error, not warn.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272641921


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +74,116 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    if (segments.isEmpty()) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
+
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from S3. See the task logs for more details."
+      );
+    }
+  }
+
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e) {
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
+      }
+      catch (AmazonServiceException e) {
+        hadException = true;
+        log.noStackTrace().warn(e,
+            "Unable to delete from bucket [%s], the following keys [%s]",
+            s3Bucket,
+            chunkOfKeys.stream().map(DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.joining(", "))
+        );
+      }
+    }
+    return hadException;
+  }
+
   @Override
   public void kill(DataSegment segment) throws SegmentLoadingException

Review Comment:
   I left that alone since we can make that change at a later date when this has been more fully vetted



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172734797


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {

Review Comment:
   make try catch block smaller



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["jasonk000 (via GitHub)" <gi...@apache.org>]

jasonk000 commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1646086691

   fyi, a related change to the SQL layer #14639


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1271017429


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -76,63 +77,114 @@ public S3DataSegmentKiller(
   @Override
   public void kill(List<DataSegment> segments) throws SegmentLoadingException
   {
-    int size = segments.size();
-    if (size == 0) {
+    if (segments.isEmpty()) {
       return;
     }
     if (segments.size() == 1) {
       kill(segments.get(0));
       return;
     }
 
-    // we can assume that all segments are in the same bucket.
-    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
-    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
 
-    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
-            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
-            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
-                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
-            .collect(Collectors.toList());
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from s3 see logs for more details"

Review Comment:
   @gianm The reason why i didn't do that is because all the information is truncated in the status for the delete task. So i figured it wasn't worth the complication to corral that state and present it in a meaningful way if it gets truncated.
   
   This is how it looks in reality in the status object.
   
   ```
   "errorMsg": "org.apache.druid.segment.loading.SegmentLoadingException: Couldn't delete from bucket: [tomfenwick-t..."
   ```
   
   that being said i can still make those changes
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1272636829


##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,29 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Kills a list of segments from deep storage. The default implementation calls kill on the segments in a loop.
+   * Implementers of this interface can leverage batch / bulk deletes to be more efficient. It is preferable to attempt
+   * to delete all segments even if there is an issue with deleting a single one. This is up to implementers to
+   * implement as putting a try catch around the default kill via iteration can be problematic if the client of the deep
+   * storage is unable to authenticate itself and segment loading exception doesn't encode enough information in it to \
+   * understand why it failed.
+   * <p>
+   * If a segment or segments does not exist in deep storage, the method should not throw an exception.
+   * <p>
+   * This version of kill must **NOT** require additional permissions on the deep storage beyond what
+   * {@link #kill(DataSegment)} requires.
+   * @param segments The list of segments to kill.
+   * @throws SegmentLoadingException If there is an exception during deletion such as a segment in the list could not be
+   * completely removed.
+   */
+  default void kill(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   I agree. This was chosen not to be done in this PR as it would require more testing



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1172736952


##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,18 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Removes segment files (indexes and metadata) from deep storage.
+   * @param segments The list of segments to kill
+   * @throws SegmentLoadingException If there is an exception during deletion
+   */
+  default void killBatched(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   come up with a better name 🤔 



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   add tests



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: WIP allow for batched delete of segments instead of deleting segment data…

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1177131874


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {
+        throw new SegmentLoadingException(e, "Couldn't kill all segment but deleted[%s]: [%s]", e.getDeletedObjects(), e);
+      }
+      catch (AmazonServiceException e) {
+

Review Comment:
   throw segmentLoadingException



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+      final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+      // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+      List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+      for (List<DataSegment> segmentsChunk : segmentsChunks) {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);

Review Comment:
   going to say no. network calls are likely the slower thing here. will add this scenario to test out before being done with this PR



##########
processing/src/main/java/org/apache/druid/segment/loading/DataSegmentKiller.java:
##########
@@ -54,6 +55,18 @@ static String descriptorPath(String path)
    */
   void kill(DataSegment segment) throws SegmentLoadingException;
 
+  /**
+   * Removes segment files (indexes and metadata) from deep storage.
+   * @param segments The list of segments to kill
+   * @throws SegmentLoadingException If there is an exception during deletion
+   */
+  default void killBatched(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   kill should be good enough



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);

Review Comment:
   this is a junk log since deleteResult doesn't implement toString ot its just a memory address :/



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +70,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    // 1000 objects is the max amount of objects that can be deleted in s3 at a time.
+    List<List<DataSegment>> segmentsChunks = Lists.partition(segments, 1000);
+    for (List<DataSegment> segmentsChunk : segmentsChunks) {
+      try {
+        DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        String[] keys = segmentsChunk.stream()
+                     .map(segment -> MapUtils.getString(segment.getLoadSpec(), "key"))
+                     .toArray(String[]::new);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(keys);
+        log.info("Removing indexes from s3, bucket: %s, keys %s", deleteObjectsRequest.getBucketName(), deleteObjectsRequest.getKeys().toString());
+        DeleteObjectsResult deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+
+        // delete descriptors which are a files to store segment metadata in deep storage.
+        // This file is deprecated and not stored anymore, but we still delete them if they exist.
+        deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+        deleteObjectsRequest = deleteObjectsRequest.withKeys(
+            Arrays.stream(keys).map(DataSegmentKiller::descriptorPath).toArray(String[]::new));
+        deleteResult = s3Client.deleteObjects(deleteObjectsRequest);
+        log.info("deleted objects %s", deleteResult);
+      }
+      catch (MultiObjectDeleteException e) {
+        throw new SegmentLoadingException(e, "Couldn't kill all segment but deleted[%s]: [%s]", e.getDeletedObjects(), e);

Review Comment:
   check these exceptions



##########
server/src/main/java/org/apache/druid/segment/loading/OmniDataSegmentKiller.java:
##########
@@ -51,6 +53,22 @@ public OmniDataSegmentKiller(
     }
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException

Review Comment:
   add test for this



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,6 +69,52 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void killBatched(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+    try {
+      // we can assume that all segments are in the same bucket.
+      String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), "bucket");

Review Comment:
   Bucket into a constant?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182824469


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,81 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    // don't fail immediately delete as many as possible then we fail in the end by throwing one exception that has all
+    // exceptions in it
+    List<Exception> exceptions = new ArrayList<>();
+    // s3 jdk is weird where the call to delete is a 200 status code for the response in the api, but the call is unable
+    // to delete some segments is an exception. this exception has a list of all objects it couldnt' delete. If you make
+    // a call that is a 401 or 403 outright the exception is thrown, but there is no list of objects it couldnt' delete.
+    // so we store them here to log later.
+    List<String> unableToDeleteKeysButNotStoredInException = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: %s the following index files: %s from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (AmazonServiceException e) {
+        // client and server errors will not have a list of keys that couldn't be deleted in the exception, so we are
+        // adding them here
+        if (e.getStatusCode() >= 300) {
+          unableToDeleteKeysButNotStoredInException.addAll(keysToDeleteStrings);
+        }
+        exceptions.add(e);
+      }
+    }
+    if (exceptions.size() > 0) {
+      List<String> segmentsNotDeleted = exceptions.stream()
+          .filter(exc -> exc instanceof MultiObjectDeleteException)
+          .map(exc -> (MultiObjectDeleteException) exc)
+          .map(MultiObjectDeleteException::getErrors)
+          .flatMap(errors -> errors.stream().map(MultiObjectDeleteException.DeleteError::getKey))
+          .collect(Collectors.toList());
+      segmentsNotDeleted.addAll(unableToDeleteKeysButNotStoredInException);
+      SegmentLoadingException segmentLoadingException =
+          new SegmentLoadingException(exceptions.get(0), "For bucket: %s unable to delete some or all segments %s", s3Bucket, segmentsNotDeleted);

Review Comment:
   For bucket: [%s] unable to delete files [%s]



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] rohangarg commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["rohangarg (via GitHub)" <gi...@apache.org>]

rohangarg commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1190191833


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);

Review Comment:
   that makes sense to me :+1:



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1650342276

   @rohangarg @maytasm I fixed the one broken test. I was told by @suneet-s and @abhishekagarwal87 to not care about the failing code coverage the default method
   
   https://github.com/apache/druid/actions/runs/5650168274/job/15306385499#step:11:2167
   https://github.com/apache/druid/actions/runs/5650168274/job/15306385132#step:11:2167


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] suneet-s commented on pull request #14131: Speed up kill tasks by deleting segments in batch

["suneet-s (via GitHub)" <gi...@apache.org>]

suneet-s commented on PR #14131:
URL: https://github.com/apache/druid/pull/14131#issuecomment-1524101171

   Added `Design Review` as this introduces a change to the `DataSegmentKiller` interface which is an extension point.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1178624962


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java:
##########
@@ -129,8 +129,12 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
     // Kill segments
     toolbox.getTaskActionClient().submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
-    for (DataSegment segment : unusedSegments) {
-      toolbox.getDataSegmentKiller().kill(segment);
+    if (getContextValue("batchDelete", false)) {

Review Comment:
   Ah i was thinking the otherway around. have it be opt in at first, then opt out. Just to make it be for the more safety conscious users to have a chance to migrate to it. Never got around to imagining of removing it entirely. Im just hung up the whole feature flag and how no other extension can take advantage of it. So it's more of a best effort feature flag.
   
   I would need to look into auto-kill and how that works



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] rohangarg commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["rohangarg (via GitHub)" <gi...@apache.org>]

rohangarg commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1188780058


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e)
+      {
+        keysNotDeleted.addAll(e.getErrors().stream()
+         .map(MultiObjectDeleteException.DeleteError::getKey)
+         .collect(Collectors.toList()));
+      }
+      catch (AmazonServiceException e)
+      {
+        throw new SegmentLoadingException(e,

Review Comment:
   might be better to add the message as something like `Couldn't completely kill segments : [%s] due to [%s]` like it is done in single `kill`



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);

Review Comment:
   can we re-use `S3Utils.deleteBucketKeys` here? 



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (MultiObjectDeleteException e)
+      {
+        keysNotDeleted.addAll(e.getErrors().stream()
+         .map(MultiObjectDeleteException.DeleteError::getKey)
+         .collect(Collectors.toList()));
+      }
+      catch (AmazonServiceException e)
+      {
+        throw new SegmentLoadingException(e,
+                                          "Unable to delete from bucket [%s]",
+                                          s3Bucket);
+      }
+    }
+    if (!keysNotDeleted.isEmpty()) {
+      throw new SegmentLoadingException("Couldn't delete from bucket: [%s] these files [%s]", s3Bucket, keysNotDeleted);

Review Comment:
   similar message comment here



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1182826484


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,81 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    deleteObjectsRequest.setQuiet(true);
+    // don't fail immediately delete as many as possible then we fail in the end by throwing one exception that has all
+    // exceptions in it
+    List<Exception> exceptions = new ArrayList<>();
+    // s3 jdk is weird where the call to delete is a 200 status code for the response in the api, but the call is unable
+    // to delete some segments is an exception. this exception has a list of all objects it couldnt' delete. If you make
+    // a call that is a 401 or 403 outright the exception is thrown, but there is no list of objects it couldnt' delete.
+    // so we store them here to log later.
+    List<String> unableToDeleteKeysButNotStoredInException = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: %s the following index files: %s from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);
+      }
+      catch (AmazonServiceException e) {
+        // client and server errors will not have a list of keys that couldn't be deleted in the exception, so we are
+        // adding them here
+        if (e.getStatusCode() >= 300) {
+          unableToDeleteKeysButNotStoredInException.addAll(keysToDeleteStrings);
+        }
+        exceptions.add(e);
+      }
+    }
+    if (exceptions.size() > 0) {
+      List<String> segmentsNotDeleted = exceptions.stream()
+          .filter(exc -> exc instanceof MultiObjectDeleteException)
+          .map(exc -> (MultiObjectDeleteException) exc)
+          .map(MultiObjectDeleteException::getErrors)
+          .flatMap(errors -> errors.stream().map(MultiObjectDeleteException.DeleteError::getKey))
+          .collect(Collectors.toList());
+      segmentsNotDeleted.addAll(unableToDeleteKeysButNotStoredInException);
+      SegmentLoadingException segmentLoadingException =
+          new SegmentLoadingException(exceptions.get(0), "For bucket: %s unable to delete some or all segments %s", s3Bucket, segmentsNotDeleted);
+      for (int ii = 1; ii < exceptions.size(); ii++) {
+        segmentLoadingException.addSuppressed(exceptions.get(ii));

Review Comment:
   what does stacktrace look like with multiple exceptions



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1245989468


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,66 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+    }
+
+    // we can assume that all segments are in the same bucket.

Review Comment:
   makes sense, I changed the code to handle this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["gianm (via GitHub)" <gi...@apache.org>]

gianm commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1263028335


##########
extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentKillerTest.java:
##########
@@ -347,7 +347,7 @@ public void test_kill_listOfSegments_multiDeleteExceptionIsThrown()
       segmentKiller.kill(ImmutableList.of(DATA_SEGMENT_1, DATA_SEGMENT_2));
     }
     catch (SegmentLoadingException exc) {
-      Assert.assertEquals("Couldn't delete from bucket: [test_bucket] these files [[key1/]]", exc.getMessage());
+      Assert.assertEquals("Couldn't delete segments from s3 see logs for more details", exc.getMessage());

Review Comment:
   You should use `Assert.assertThrows` rather than doing this. This approach has the problem that if `segmentKiller.kill` _doesn't_ throw an exception for some reason, then the test will still pass, because this `Assert.assertEquals` never gets called. But, we want it to fail if the expected exception isn't thrown.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -76,63 +77,114 @@ public S3DataSegmentKiller(
   @Override
   public void kill(List<DataSegment> segments) throws SegmentLoadingException
   {
-    int size = segments.size();
-    if (size == 0) {
+    if (segments.isEmpty()) {
       return;
     }
     if (segments.size() == 1) {
       kill(segments.get(0));
       return;
     }
 
-    // we can assume that all segments are in the same bucket.
-    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
-    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
 
-    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
-            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
-            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
-                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
-            .collect(Collectors.toList());
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from s3 see logs for more details"
+      );
+    }
+  }
 
-    // max delete object request size is 1000 for S3
-    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
     DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
-    // only return objects failed to delete.
     deleteObjectsRequest.setQuiet(true);
-
-    List<String> keysNotDeleted = new ArrayList<>();
-    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
-      List<String> keysToDeleteStrings = keysChunk.stream().map(
-            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
       try {
-        deleteObjectsRequest.setKeys(keysChunk);
-        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
         s3Client.deleteObjects(deleteObjectsRequest);
       }
       catch (MultiObjectDeleteException e) {
-        keysNotDeleted.addAll(e.getErrors().stream()
-                               .map(MultiObjectDeleteException.DeleteError::getKey)
-                               .collect(Collectors.toList()));
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
       }
       catch (AmazonServiceException e) {
-        throw new SegmentLoadingException(e,
-                                          "Unable to delete from bucket [%s]",
-                                          s3Bucket);
+        hadException = true;
+        log.error(

Review Comment:
   When logging it's best to put `e` in the first position rather than interpolate `e.getMessage()`. If your goal is to avoid showing a stack trace, you can do `log.noStackTrace().error(...)`. This is nice because if people enable debug-level logging, the stack traces are shown anyway, providing admins with a way to get the stack traces if needed.
   
   Also, a `warn` would be better here. Generally `log.error` should be reserved for stuff that is really serious and affects the entire service.



##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -76,63 +77,114 @@ public S3DataSegmentKiller(
   @Override
   public void kill(List<DataSegment> segments) throws SegmentLoadingException
   {
-    int size = segments.size();
-    if (size == 0) {
+    if (segments.isEmpty()) {
       return;
     }
     if (segments.size() == 1) {
       kill(segments.get(0));
       return;
     }
 
-    // we can assume that all segments are in the same bucket.
-    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
-    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
 
-    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
-            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
-            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
-                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
-            .collect(Collectors.toList());
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from s3 see logs for more details"

Review Comment:
   Please improve the grammar (missing punctuation) and capitalization (S3).
   
   Also, can we interpolate some useful information into this error message? Such as the number of segments, the first segment ID (or first N IDs) that couldn't be deleted, the nature of the reason why it couldn't be deleted, etc. This info could come back from `deleteKeysForBucket` rather than a `boolean`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Speed up kill tasks by deleting segments in batch (druid)

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1263042438


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -76,63 +77,114 @@ public S3DataSegmentKiller(
   @Override
   public void kill(List<DataSegment> segments) throws SegmentLoadingException
   {
-    int size = segments.size();
-    if (size == 0) {
+    if (segments.isEmpty()) {
       return;
     }
     if (segments.size() == 1) {
       kill(segments.get(0));
       return;
     }
 
-    // we can assume that all segments are in the same bucket.
-    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
-    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    // create a map of bucket to keys to delete
+    Map<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeysToDelete = new HashMap<>();
+    for (DataSegment segment : segments) {
+      String s3Bucket = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.BUCKET);
+      String path = MapUtils.getString(segment.getLoadSpec(), S3DataSegmentPuller.KEY);
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(path));
+      keysToDelete.add(new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path)));
+      bucketToKeysToDelete.computeIfAbsent(s3Bucket, k -> new ArrayList<>()).addAll(keysToDelete);
+    }
 
-    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
-            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
-            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
-                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
-            .collect(Collectors.toList());
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+    boolean shouldThrowException = false;
+    for (Map.Entry<String, List<DeleteObjectsRequest.KeyVersion>> bucketToKeys : bucketToKeysToDelete.entrySet()) {
+      String s3Bucket = bucketToKeys.getKey();
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete = bucketToKeys.getValue();
+      boolean hadException = deleteKeysForBucket(s3Client, s3Bucket, keysToDelete);
+      if (hadException) {
+        shouldThrowException = true;
+      }
+    }
+    if (shouldThrowException) {
+      // exception error message gets cutoff without providing any details. look at the logs for more details.
+      // this was a shortcut to handle the many different ways there could potentially be failures and handle them
+      // reasonably
+      throw new SegmentLoadingException(
+          "Couldn't delete segments from s3 see logs for more details"
+      );
+    }
+  }
 
-    // max delete object request size is 1000 for S3
-    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+  /**
+   * Delete all keys in a bucket from s3
+   *
+   * @param s3Client     client used to communicate with s3
+   * @param s3Bucket     the bucket where the keys exist
+   * @param keysToDelete the keys to delete
+   * @return a boolean value of true if there was an issue deleting one or many keys, a boolean value of false if
+   * succesful
+   */
+  private boolean deleteKeysForBucket(
+      ServerSideEncryptingAmazonS3 s3Client,
+      String s3Bucket,
+      List<DeleteObjectsRequest.KeyVersion> keysToDelete
+  )
+  {
+    boolean hadException = false;
     DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
-    // only return objects failed to delete.
     deleteObjectsRequest.setQuiet(true);
-
-    List<String> keysNotDeleted = new ArrayList<>();
-    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
-      List<String> keysToDeleteStrings = keysChunk.stream().map(
-            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(
+        keysToDelete,
+        MAX_MULTI_OBJECT_DELETE_SIZE
+    );
+    for (List<DeleteObjectsRequest.KeyVersion> chunkOfKeys : keysChunks) {
+      List<String> keysToDeleteStrings = chunkOfKeys.stream().map(
+          DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
       try {
-        deleteObjectsRequest.setKeys(keysChunk);
-        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        deleteObjectsRequest.setKeys(chunkOfKeys);
+        log.info(
+            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            s3Bucket,
+            keysToDeleteStrings
+        );
         s3Client.deleteObjects(deleteObjectsRequest);
       }
       catch (MultiObjectDeleteException e) {
-        keysNotDeleted.addAll(e.getErrors().stream()
-                               .map(MultiObjectDeleteException.DeleteError::getKey)
-                               .collect(Collectors.toList()));
+        hadException = true;
+        Map<String, List<String>> errorToKeys = new HashMap<>();
+        for (MultiObjectDeleteException.DeleteError error : e.getErrors()) {
+          error.getMessage();
+          error.getKey();
+          errorToKeys.computeIfAbsent(error.getMessage(), k -> new ArrayList<>()).add(error.getKey());
+        }
+        errorToKeys.forEach((key, value) -> log.error(
+            "Unable to delete from bucket [%s], the following keys [%s], because [%s]",
+            s3Bucket,
+            String.join(", ", value),
+            key
+        ));
       }
       catch (AmazonServiceException e) {
-        throw new SegmentLoadingException(e,
-                                          "Unable to delete from bucket [%s]",
-                                          s3Bucket);
+        hadException = true;
+        log.error(

Review Comment:
   will do



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] TSFenwick commented on a diff in pull request #14131: Speed up kill tasks by deleting segments in batch

["TSFenwick (via GitHub)" <gi...@apache.org>]

TSFenwick commented on code in PR #14131:
URL: https://github.com/apache/druid/pull/14131#discussion_r1212049428


##########
extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java:
##########
@@ -64,13 +73,69 @@ public S3DataSegmentKiller(
     this.inputDataConfig = inputDataConfig;
   }
 
+  @Override
+  public void kill(List<DataSegment> segments) throws SegmentLoadingException
+  {
+    int size = segments.size();
+    if (size == 0) {
+      return;
+    }
+    if (segments.size() == 1) {
+      kill(segments.get(0));
+      return;
+
+    }
+
+    // we can assume that all segments are in the same bucket.
+    String s3Bucket = MapUtils.getString(segments.get(0).getLoadSpec(), BUCKET);
+    final ServerSideEncryptingAmazonS3 s3Client = this.s3ClientSupplier.get();
+
+    List<DeleteObjectsRequest.KeyVersion> keysToDelete = segments.stream()
+            .map(segment -> MapUtils.getString(segment.getLoadSpec(), KEY))
+            .flatMap(path -> Stream.of(new DeleteObjectsRequest.KeyVersion(path),
+                                     new DeleteObjectsRequest.KeyVersion(DataSegmentKiller.descriptorPath(path))))
+            .collect(Collectors.toList());
+
+    // max delete object request size is 1000 for S3
+    List<List<DeleteObjectsRequest.KeyVersion>> keysChunks = Lists.partition(keysToDelete, 1000);
+    DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(s3Bucket);
+    // only return objects failed to delete.
+    deleteObjectsRequest.setQuiet(true);
+
+    List<String> keysNotDeleted = new ArrayList<>();
+    for (List<DeleteObjectsRequest.KeyVersion> keysChunk : keysChunks) {
+      List<String> keysToDeleteStrings = keysChunk.stream().map(
+            DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      try {
+        deleteObjectsRequest.setKeys(keysChunk);
+        log.info("Removing from bucket: [%s] the following index files: [%s] from s3!", s3Bucket, keysToDeleteStrings);
+        s3Client.deleteObjects(deleteObjectsRequest);

Review Comment:
   @rohangarg i think the current state of this pr is good enough as it is since it is a major improvement to how it used to run. I dont want perfect to be the enemy of better. So i would prefer to get this merged and then come back to this later to make the suggested improvements



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org