Add a configurable bufferPeriod between when a segment is marked unused and deleted by KillUnusedSegments duty

core/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java

@@ -88,4 +88,22 @@ default void exportTable(
   void createSupervisorsTable();
 
   void deleteAllRecords(String tableName);
+
+  /**
+   * Upgrade Compatability Method.
+   *
+   * A new column, used_flag_last_updated, is added to druid_segmens table. This method alters the table to add the column to make
+   * a cluster's metastore tables compatible with the updated Druid codebase in 0.24.x+
+   */
+  void alterSegmentTableAddUsedFlagLastUpdated();
+
+  /**
+   * Upgrade Compatability Method.
+   *
+   * Updates the druid_segments table and populates the new used_flag_last_updated column for unused segments. This makes unused
+   * segments compatible with the new 0.24.x+ code for allowing a minimum buffer period between a segment becoming
+   * unused and being deleted
+   */
+  void updateSegmentTablePopulateUsedFlagLastUpdated();
+
 }

docs/configuration/index.md

@@ -816,6 +816,7 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.kill.period`|How often to send kill tasks to the indexing service. Value must be greater than `druid.coordinator.period.indexingPeriod`. Only applies if kill is turned on.|P1D (1 Day)|
 |`druid.coordinator.kill.durationToRetain`|Only applies if you set `druid.coordinator.kill.on` to `true`. This value is ignored if `druid.coordinator.kill.ignoreDurationToRetain` is `true`. Valid configurations must be a ISO8601 period. Druid will not kill unused segments whose interval end date is beyond `now - durationToRetain`. `durationToRetain` can be a negative ISO8601 period, which would result in `now - durationToRetain` to be in the future.<br /><br />Note that the `durationToRetain` parameter applies to the segment interval, not the time that the segment was last marked unused. For example, if `durationToRetain` is set to `P90D`, then a segment for a time chunk 90 days in the past is eligible for permanent deletion immediately after being marked unused.|`P90D`|
 |`druid.coordinator.kill.ignoreDurationToRetain`|A way to override `druid.coordinator.kill.durationToRetain` and tell the coordinator that you do not care about the end date of unused segment intervals when it comes to killing them. If true, the coordinator considers all unused segments as eligible to be killed.|false|
+|`druid.coordinator.kill.bufferPeriod`|The amount of time that a segment must be unused before it is able to be permanently removed from metadata and deep storage. This can serve as a buffer period to prevent data loss if data ends up being needed after being marked unused.|`PT24H`|
 |`druid.coordinator.kill.maxSegments`|Kill at most n unused segments per kill task submission, must be greater than 0. Only applies and MUST be specified if kill is turned on.|100|
 |`druid.coordinator.balancer.strategy`|Specify the type of balancing strategy for the coordinator to use to distribute segments among the historicals. `cachingCost` is logically equivalent to `cost` but is more CPU-efficient on large clusters. `diskNormalized` weights the costs according to the servers' disk usage ratios - there are known issues with this strategy distributing segments unevenly across the cluster. `random` distributes segments among services randomly.|`cost`|
 |`druid.coordinator.balancer.cachingCost.awaitInitialization`|Whether to wait for segment view initialization before creating the `cachingCost` balancing strategy. This property is enabled only when `druid.coordinator.balancer.strategy` is `cachingCost`. If set to 'true', the Coordinator will not start to assign segments, until the segment view is initialized. If set to 'false', the Coordinator will fallback to use the `cost` balancing strategy only if the segment view is not initialized yet. Notes, it may take much time to wait for the initialization since the `cachingCost` balancing strategy involves much computing to build itself.|false|

docs/dependencies/metadata-storage.md

@@ -99,7 +99,9 @@ system. The table has two main functional columns, the other columns are for ind
 Value 1 in the `used` column means that the segment should be "used" by the cluster (i.e., it should be loaded and
 available for requests). Value 0 means that the segment should not be loaded into the cluster. We do this as a means of
 unloading segments from the cluster without actually removing their metadata (which allows for simpler rolling back if
-that is ever an issue).
+that is ever an issue). The `used` column has a corresponding `used_flag_last_updated` column that indicates the date at the instant
+that the `used` status of the segment was last updated. This information can be used by the coordinator to determine if
+a segment is a candidate for deletion (if automated segment killing is enabled).
 
 The `payload` column stores a JSON blob that has all of the metadata for the segment.
 Some of the data in the `payload` column intentionally duplicates data from other columns in the segments table.

docs/operations/upgrade-prep.md

@@ -0,0 +1,114 @@
+---
+id: upgrade-prep
+title: "Upgrade Prep"
+---
+
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one
+  ~ or more contributor license agreements.  See the NOTICE file
+  ~ distributed with this work for additional information
+  ~ regarding copyright ownership.  The ASF licenses this file
+  ~ to you under the Apache License, Version 2.0 (the
+  ~ "License"); you may not use this file except in compliance
+  ~ with the License.  You may obtain a copy of the License at
+  ~
+  ~   http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing,
+  ~ software distributed under the License is distributed on an
+  ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  ~ KIND, either express or implied.  See the License for the
+  ~ specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+## Upgrade to `0.24+` from `0.23` and earlier
+
+### Altering segments table
+
+**If you have set `druid.metadata.storage.connector.createTables` to `true` (which is the default), and your metadata connect user has DDL privileges, you can disregard this section. You are urged to still evaluate the optional section below**
+
+**The coordinator and overlord services will fail if you do not execute this change prior to the upgrade**
+
+A new column, `used_flag_last_updated`, is needed in the segments table to support new
+segment killing functionality. You can manually alter the table, or you can use
+a CLI tool to perform the update.
+
+#### CLI tool
+
+Druid provides a `metadata-update` tool for updating Druid's metadata tables.
+
+In the example commands below:
+
+- `lib` is the Druid lib directory
+- `extensions` is the Druid extensions directory
+- `base` corresponds to the value of `druid.metadata.storage.tables.base` in the configuration, `druid` by default.
+- The `--connectURI` parameter corresponds to the value of `druid.metadata.storage.connector.connectURI`.
+- The `--user` parameter corresponds to the value of `druid.metadata.storage.connector.user`.
+- The `--password` parameter corresponds to the value of `druid.metadata.storage.connector.password`.
+- The `--action` parameter corresponds to the update action you are executing. In this case it is: `add-last-used-to-segments`
+
+##### MySQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"mysql-metadata-storage\"] -Ddruid.metadata.storage.type=mysql org.apache.druid.cli.Main tools metadata-update --connectURI="<mysql-uri>" --user <user> --password <pass> --base druid --action add-used-flag-last-updated-to-segments
+```
+
+##### PostgreSQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"postgresql-metadata-storage\"] -Ddruid.metadata.storage.type=postgresql org.apache.druid.cli.Main tools metadata-update --connectURI="<postgresql-uri>" --user <user> --password <pass> --base druid --action add-used-flag-last-updated-to-segments
+```
+
+
+#### Manual ALTER TABLE
+
+```SQL
+ALTER TABLE druid_segments
+ADD used_flag_last_updated varchar(255);
+```
+
+### Populating `used_flag_last_updated` column of the segments table after upgrade (Optional)
+
+This is an optional step to take **after** you upgrade the Overlord and Coordinator to `0.24+` (from `0.23` and earlier). If you do not take this action and are also using `druid.coordinator.kill.on=true`, the logic to identify segments that can be killed will not honor `druid.coordinator.kill.bufferPeriod` for the rows in the segments table where `used_flag_last_updated == null`.
+
+#### CLI tool
+
+Druid provides a `metadata-update` tool for updating Druid's metadata tables. Note that this tool will update `used_flag_last_updated` for all rows that match `used = false` in one transaction.
+
+In the example commands below:
+
+- `lib` is the Druid lib directory
+- `extensions` is the Druid extensions directory
+- `base` corresponds to the value of `druid.metadata.storage.tables.base` in the configuration, `druid` by default.
+- The `--connectURI` parameter corresponds to the value of `druid.metadata.storage.connector.connectURI`.
+- The `--user` parameter corresponds to the value of `druid.metadata.storage.connector.user`.
+- The `--password` parameter corresponds to the value of `druid.metadata.storage.connector.password`.
+- The `--action` parameter corresponds to the update action you are executing. In this case it is: `add-last-used-to-segments`
+
+##### MySQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"mysql-metadata-storage\"] -Ddruid.metadata.storage.type=mysql org.apache.druid.cli.Main tools metadata-update --connectURI="<mysql-uri>" --user <user> --password <pass> --base druid --action populate-used-flag-last-updated-column-in-segments
+```
+
+##### PostgreSQL
+
+```bash
+cd ${DRUID_ROOT}
+java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[\"postgresql-metadata-storage\"] -Ddruid.metadata.storage.type=postgresql org.apache.druid.cli.Main tools metadata-update --connectURI="<postgresql-uri>" --user <user> --password <pass> --base druid --action populate-used-flag-last-updated-column-in-segments
+```
+
+
+#### Manual UPDATE
+
+Note that we choose a random date string for this example. We recommend using the current UTC time when you invoke the command. If you have lots of rows that match the conditional `used = false`, you may want to incrementally update the table using a limit clause.
+
+```SQL
+UPDATE druid_segment
+SET used_flag_last_updated = '2022-01-01T00:00:00.000Z'
+where used = false;
+```

pom.xml

@@ -1185,6 +1185,9 @@
                         <exclude>org/apache/druid/server/initialization/jetty/*Module*</exclude>
                         <exclude>org/apache/druid/guice/http/*</exclude>
 
+                        <!-- Ignore cli tools -->
+                        <exclude>org/apache/druid/cli/UpdateTables*</exclude>
+
                         <!-- Ignore generated code -->
                         <exclude>org/apache/druid/math/expr/antlr/Expr*</exclude>  <!-- core -->
                         <exclude>org/apache/druid/**/generated/*Benchmark*</exclude>  <!-- benchmarks -->

server/src/main/java/org/apache/druid/indexer/SQLMetadataStorageUpdaterJobHandler.java

@@ -59,8 +59,8 @@ public Void withHandle(Handle handle) throws Exception
           {
             final PreparedBatch batch = handle.prepareBatch(
                 StringUtils.format(
-                    "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload) "
-                    + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload)",
+                    "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload, used_flag_last_updated) "
+                    + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload, :used_flag_last_updated)",
                     tableName, connector.getQuoteString()
                 )
             );
@@ -77,6 +77,7 @@ public Void withHandle(Handle handle) throws Exception
                       .put("version", segment.getVersion())
                       .put("used", true)
                       .put("payload", mapper.writeValueAsBytes(segment))
+                      .put("used_flag_last_updated", DateTimes.nowUtc().toString())
                       .build()
               );
 

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

@@ -1006,8 +1006,8 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
 
       PreparedBatch preparedBatch = handle.prepareBatch(
           StringUtils.format(
-              "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload) "
-                  + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload)",
+              "INSERT INTO %1$s (id, dataSource, created_date, start, %2$send%2$s, partitioned, version, used, payload, used_flag_last_updated) "
+                  + "VALUES (:id, :dataSource, :created_date, :start, :end, :partitioned, :version, :used, :payload, :used_flag_last_updated)",
               dbTables.getSegmentsTable(),
               connector.getQuoteString()
           )
@@ -1024,7 +1024,8 @@ private Set<DataSegment> announceHistoricalSegmentBatch(
               .bind("partitioned", (segment.getShardSpec() instanceof NoneShardSpec) ? false : true)
               .bind("version", segment.getVersion())
               .bind("used", usedSegments.contains(segment))
-              .bind("payload", jsonMapper.writeValueAsBytes(segment));
+              .bind("payload", jsonMapper.writeValueAsBytes(segment))
+              .bind("used_flag_last_updated", DateTimes.nowUtc().toString());
         }
         final int[] affectedRows = preparedBatch.execute();
         final boolean succeeded = Arrays.stream(affectedRows).allMatch(eachAffectedRows -> eachAffectedRows == 1);