@Override
    public void insert(Metric metric) throws MetricException {
        try {
            // don't bother blocking on a full queue, just drop metrics in case we can't keep up
            if (queue.remainingCapacity() <= 0) {
                LOG.info("Metrics q full, dropping metric");
                return;
            }
            queue.put(metric);
        } catch (Exception e) {
            String message = "Failed to insert metric";
            LOG.error(message, e);
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException(message, e);
        }
    }
@Test
    public void testMetricCleanup() throws Exception {
        FilterOptions filter;
        List<Metric> list;

        // Share some common metadata strings to validate they do not get deleted
        String commonTopologyId = "topology-cleanup-2";
        String commonStreamId = "stream-cleanup-5";
        String defaultS = "default";
        Metric m1 = new Metric(defaultS, 40000000L, commonTopologyId, 1.0,
                "component-1", defaultS, "hostname-1", commonStreamId, 1, AggLevel.AGG_LEVEL_NONE);
        Metric m2 = new Metric(defaultS, System.currentTimeMillis(), commonTopologyId, 1.0,
                "component-1", "executor-1", defaultS, commonStreamId, 1, AggLevel.AGG_LEVEL_NONE);

        store.insert(m1);
        store.insert(m2);
        waitForInsertFinish(m2);

        // validate at least two agg level none metrics exist
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        list = getMetricsFromScan(filter);
        assertTrue(list.size() >= 2);

        // delete anything older than an hour
        MetricsCleaner cleaner = new MetricsCleaner((RocksDbStore)store, 1, 1, null, new StormMetricsRegistry());
        cleaner.purgeMetrics();
        list = getMetricsFromScan(filter);
        assertEquals(1, list.size());
        assertTrue(list.contains(m2));
    }