@Override
    public void scan(FilterOptions filter, ScanCallback scanCallback) throws MetricException {
        scanInternal(filter, scanCallback, null);
    }
@Test
    public void testScan() throws Exception {
        FilterOptions filter;
        List<Metric> list;

        Metric m1 = new Metric("metricType1", 50000000L, "Topo-m1", 1.0,
                "component-1", "executor-2", "hostname-1", "stream-1", 1, AggLevel.AGG_LEVEL_NONE);
        Metric m2 = new Metric("metricType2", 50030000L, "Topo-m1", 1.0,
                "component-1", "executor-1", "hostname-2", "stream-2", 1, AggLevel.AGG_LEVEL_NONE);
        Metric m3 = new Metric("metricType3", 50200000L, "Topo-m1", 1.0,
                "component-2", "executor-1", "hostname-1", "stream-3", 1, AggLevel.AGG_LEVEL_NONE);
        Metric m4 = new Metric("metricType4", 50200000L, "Topo-m2", 1.0,
                "component-2", "executor-1", "hostname-2", "stream-4", 2, AggLevel.AGG_LEVEL_NONE);
        store.insert(m1);
        store.insert(m2);
        store.insert(m3);
        store.insert(m4);
        waitForInsertFinish(m4);

        // validate search by time
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setStartTime(50000000L);
        filter.setEndTime(50130000L);
        list = getMetricsFromScan(filter);
        assertEquals(2, list.size());
        assertTrue(list.contains(m1));
        assertTrue(list.contains(m2));

        // validate search by topology id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setTopologyId("Topo-m2");
        list = getMetricsFromScan(filter);
        assertEquals(1, list.size());
        assertTrue(list.contains(m4));

        // validate search by metric id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setMetricName("metricType2");
        list = getMetricsFromScan(filter);
        assertEquals(1, list.size());
        assertTrue(list.contains(m2));

        // validate search by component id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setComponentId("component-2");
        list = getMetricsFromScan(filter);
        assertEquals(2, list.size());
        assertTrue(list.contains(m3));
        assertTrue(list.contains(m4));

        // validate search by executor id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setExecutorId("executor-1");
        list = getMetricsFromScan(filter);
        assertEquals(3, list.size());
        assertTrue(list.contains(m2));
        assertTrue(list.contains(m3));
        assertTrue(list.contains(m4));

        // validate search by executor id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setExecutorId("executor-1");
        list = getMetricsFromScan(filter);
        assertEquals(3, list.size());
        assertTrue(list.contains(m2));
        assertTrue(list.contains(m3));
        assertTrue(list.contains(m4));

        // validate search by host id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setHostId("hostname-2");
        list = getMetricsFromScan(filter);
        assertEquals(2, list.size());
        assertTrue(list.contains(m2));
        assertTrue(list.contains(m4));

        // validate search by port
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setPort(1);
        list = getMetricsFromScan(filter);
        assertEquals(3, list.size());
        assertTrue(list.contains(m1));
        assertTrue(list.contains(m2));
        assertTrue(list.contains(m3));

        // validate search by stream id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setStreamId("stream-4");
        list = getMetricsFromScan(filter);
        assertEquals(1, list.size());
        assertTrue(list.contains(m4));

        // validate 4 metrics (aggregations) found for m4 for all agglevels when searching by port
        filter = new FilterOptions();
        filter.setPort(2);
        list = getMetricsFromScan(filter);
        assertEquals(4, list.size());
        assertTrue(list.contains(m4));
        assertFalse(list.contains(m1));
        assertFalse(list.contains(m2));
        assertFalse(list.contains(m3));

        // validate search by topology id and executor id
        filter = new FilterOptions();
        filter.addAggLevel(AggLevel.AGG_LEVEL_NONE);
        filter.setTopologyId("Topo-m1");
        filter.setExecutorId("executor-1");
        list = getMetricsFromScan(filter);
        assertEquals(2, list.size());
        assertTrue(list.contains(m2));
        assertTrue(list.contains(m3));
    }