@SuppressWarnings("rawtypes")
  public Set<Object> gcTombstones(LocalRegion r, Map<VersionSource<?>, Long> regionGCVersions,
      boolean needsKeys) {
    synchronized (getBlockGCLock()) {
      int count = getGCBlockCount();
      if (count > 0) {
        // if any delta GII is on going as provider at this member, not to do tombstone GC
        if (logger.isDebugEnabled()) {
          logger.debug("gcTombstones skipped due to {} Delta GII on going", count);
        }
        return null;
      }
      if (logger.isDebugEnabled()) {
        logger.debug("gcTombstones invoked for region {} and version map {}", r, regionGCVersions);
      }
      final VersionSource myId = r.getVersionMember();
      final TombstoneSweeper sweeper = getSweeper(r);
      final List<Tombstone> removals = new ArrayList<>();
      sweeper.removeUnexpiredIf(t -> {
        if (t.region == r) {
          VersionSource destroyingMember = t.getMemberID();
          if (destroyingMember == null) {
            destroyingMember = myId;
          }
          Long maxReclaimedRV = regionGCVersions.get(destroyingMember);
          if (maxReclaimedRV != null && t.getRegionVersion() <= maxReclaimedRV) {
            removals.add(t);
            return true;
          }
        }
        return false;
      });

      // Record the GC versions now, so that we can persist them
      for (Map.Entry<VersionSource<?>, Long> entry : regionGCVersions.entrySet()) {
        r.getVersionVector().recordGCVersion(entry.getKey(), entry.getValue());
      }

      // Remove any exceptions from the RVV that are older than the GC version
      r.getVersionVector().pruneOldExceptions();

      // Persist the GC RVV to disk. This needs to happen BEFORE we remove
      // the entries from map, to prevent us from removing a tombstone
      // from disk that has a version greater than the persisted
      // GV RVV.
      if (r.getDataPolicy().withPersistence()) {
        // Update the version vector which reflects what has been persisted on disk.
        r.getDiskRegion().writeRVVGC(r);
      }

      Set<Object> removedKeys = needsKeys ? new HashSet<>() : Collections.emptySet();
      for (Tombstone t : removals) {
        boolean tombstoneWasStillInRegionMap = t.region.getRegionMap().removeTombstone(t.entry, t);
        if (needsKeys && tombstoneWasStillInRegionMap) {
          removedKeys.add(t.entry.getKey());
        }
      }
      return removedKeys;
    } // sync on deltaGIILock
  }
remove tombstones from the given region that have region-versions <= those in the given removal
map

@return a collection of keys removed (only if the region is a bucket - empty otherwise)
geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java