docs: public void scheduleTTLTasks() {
  
          //do not trigger TTL on replica clusters
          RealVoltDB db = (RealVoltDB)VoltDB.instance();
          if (db.getReplicationRole() == ReplicationRole.REPLICA) {
              shutDown();
              return;
          }
  
          Map<String, Table> ttlTables = CatalogUtil.getTimeToLiveTables(db.m_catalogContext.database);
          if (m_timeToLiveExecutor == null && !ttlTables.isEmpty()) {
              m_timeToLiveExecutor = CoreUtils.getScheduledThreadPoolExecutor("TimeToLive", 1, CoreUtils.SMALL_STACK_SIZE);
              m_timeToLiveExecutor.setRemoveOnCancelPolicy(true);
          }
  
          //remove dropped TTL tasks
          String info = "TTL task for table %s";
          Iterator<Map.Entry<String, TTLTask>> it = m_tasks.entrySet().iterator();
          while(it.hasNext()) {
              Map.Entry<String, TTLTask> task = it.next();
              if (!ttlTables.containsKey(task.getKey())) {
                  task.getValue().cancel();
                  it.remove();
                  m_stats.remove(task.getKey());
                  hostLog.info(String.format(info + " has been dropped.", task.getKey()));
              }
          }
  
          //random initial delay
          final Random random = new Random();
          for (Table t : ttlTables.values()) {
              TimeToLive ttl = t.getTimetolive().get(TimeToLiveVoltDB.TTL_NAME);
              if (!CatalogUtil.isColumnIndexed(t, ttl.getTtlcolumn())) {
                  hostLog.warn("An index is missing on column " + t.getTypeName() + "." + ttl.getTtlcolumn().getName() +
                          " for TTL. No records will be purged until an index is added.");
                  continue;
              }
              TTLTask task = m_tasks.get(t.getTypeName());
              if (task == null) {
                  TTLStats stats = m_stats.get(t.getTypeName());
                  if (!TableType.isPersistentMigrate(t.getTabletype()) && stats == null) {
                      stats = new TTLStats(t.getTypeName());
                      m_stats.put(t.getTypeName(), stats);
                  }
                  task = new TTLTask(t.getTypeName(), ttl, stats);
                  m_tasks.put(t.getTypeName(), task);
                  m_futures.put(t.getTypeName(),
                                m_timeToLiveExecutor.scheduleAtFixedRate(task,
                                        DELAY + random.nextInt(INTERVAL),
                                        INTERVAL, TimeUnit.MILLISECONDS));
                  hostLog.info(String.format(info + " has been scheduled.", t.getTypeName()));
              } else {
                  task.updateTask(ttl);
                  hostLog.info(String.format(info + " has been updated.", t.getTypeName()));
              }
          }
      }