@Override
  protected void periodicExecute(final TEnvironment env) {
    if (completed.isEmpty()) {
      if (LOG.isTraceEnabled()) {
        LOG.trace("No completed procedures to cleanup.");
      }
      return;
    }

    final long evictTtl =
      conf.getInt(ProcedureExecutor.EVICT_TTL_CONF_KEY, ProcedureExecutor.DEFAULT_EVICT_TTL);
    final long evictAckTtl = conf.getInt(ProcedureExecutor.EVICT_ACKED_TTL_CONF_KEY,
      ProcedureExecutor.DEFAULT_ACKED_EVICT_TTL);
    final int batchSize = conf.getInt(BATCH_SIZE_CONF_KEY, DEFAULT_BATCH_SIZE);

    final long[] batchIds = new long[batchSize];
    int batchCount = 0;

    final long now = EnvironmentEdgeManager.currentTime();
    final Iterator<Map.Entry<Long, CompletedProcedureRetainer<TEnvironment>>> it =
      completed.entrySet().iterator();
    while (it.hasNext() && store.isRunning()) {
      final Map.Entry<Long, CompletedProcedureRetainer<TEnvironment>> entry = it.next();
      final CompletedProcedureRetainer<TEnvironment> retainer = entry.getValue();
      final Procedure<?> proc = retainer.getProcedure();
      IdLock.Entry lockEntry;
      try {
        lockEntry = procExecutionLock.getLockEntry(proc.getProcId());
      } catch (IOException e) {
        // can only happen if interrupted, so not a big deal to propagate it
        throw new UncheckedIOException(e);
      }
      try {
        // TODO: Select TTL based on Procedure type
        if (retainer.isExpired(now, evictTtl, evictAckTtl)) {
          // Failed procedures aren't persisted in WAL.
          if (!(proc instanceof FailedProcedure)) {
            batchIds[batchCount++] = entry.getKey();
            if (batchCount == batchIds.length) {
              store.delete(batchIds, 0, batchCount);
              batchCount = 0;
            }
          }
          final NonceKey nonceKey = proc.getNonceKey();
          if (nonceKey != null) {
            nonceKeysToProcIdsMap.remove(nonceKey);
          }
          it.remove();
          LOG.trace("Evict completed {}", proc);
        }
      } finally {
        procExecutionLock.releaseLockEntry(lockEntry);
      }
    }
    if (batchCount > 0) {
      store.delete(batchIds, 0, batchCount);
    }
    // let the store do some cleanup works, i.e, delete the place marker for preserving the max
    // procedure id.
    store.cleanup();
  }
Internal cleaner that removes the completed procedure results after a TTL.
<p/>
NOTE: This is a special case handled in timeoutLoop().
<p/>
Since the client code looks more or less like:

<pre>
  procId = master.doOperation()
  while (master.getProcResult(procId) == ProcInProgress);
</pre>

The master should not throw away the proc result as soon as the procedure is done but should wait
a result request from the client (see executor.removeResult(procId)) The client will call
something like master.isProcDone() or master.getProcResult() which will return the result/state
to the client, and it will mark the completed proc as ready to delete. note that the client may
not receive the response from the master (e.g. master failover) so, if we delay a bit the real
deletion of the proc result the client will be able to get the result the next try.
hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/CompletedProcedureCleaner.java