@Test
public void testReservoirSampler() throws IOException, InterruptedException {
Benchmark b = new Benchmark("Reservoir sampler + nullsink");
b.mark("begin");
MemorySinkSource mem = FlumeBenchmarkHarness.synthInMem();
b.mark("disk_loaded");
NullSink nullsnk = new NullSink();
EventUtil.dumpAll(mem, nullsnk);
b.mark("warmup");
int res_size[] = { 100, 1000, 10000, 100000 };
for (int i = 0; i < res_size.length; i++) {
mem.open();
int sz = res_size[i];
EventSink res = new ReservoirSamplerDeco<NullSink>(new NullSink(), sz);
EventUtil.dumpAll(mem, res);
b.mark("reservoir " + sz + " sampling done", sz);
res.close();
b.mark("sample dump done");
}
for (int i = 0; i < res_size.length; i++) {
mem.open();
int sz = res_size[i];
CounterSink cnt = new CounterSink("null");
EventSink res = new ReservoirSamplerDeco<CounterSink>(cnt, sz);
EventUtil.dumpAll(mem, res);
b.mark("reservoir", sz);