@Test public void testWriter() throws Exception { Counter bufferedMetrics = mock(Counter.class); IngestionContext context = mock(IngestionContext.class); AbstractMetricsRW basicRW = mock(AbstractMetricsRW.class); AbstractMetricsRW preAggrRW = mock(AbstractMetricsRW.class); MetricsRWDelegator metricsRWDelegator = new MetricsRWDelegator(basicRW, preAggrRW); List<List<IMetric>> testdata = createTestData(Metric.class); List<List<IMetric>> pTestdata = createTestData(PreaggregatedMetric.class); List<List<IMetric>> allTestdata = new ArrayList<List<IMetric>>(); allTestdata.addAll(testdata); allTestdata.addAll(pTestdata); BatchWriter batchWriter = new BatchWriter( new ThreadPoolBuilder().build(), timeout, bufferedMetrics, context, metricsRWDelegator ); ListenableFuture<List<Boolean>> futures = batchWriter.apply(allTestdata); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); Assert.assertTrue(persisteds.size() == (NUM_LISTS * 2)); for (Boolean batchStatus : persisteds) { Assert.assertTrue(batchStatus); } for (List<IMetric> l : testdata) { verify(basicRW).insertMetrics((List<IMetric>)(List<?>)l); } for (List<IMetric> l : pTestdata) { verify(preAggrRW).insertMetrics(l); } for (List<IMetric> l : allTestdata) { Assert.assertTrue(l.size() == METRICS_PER_LIST); for (IMetric m : l) { verify(context).update(m.getCollectionTime(), Util.getShard(m.getLocator().toString())); } } }
@Override public ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input) throws Exception { final long writeStartTime = System.currentTimeMillis(); final Timer.Context actualWriteCtx = writeDurationTimer.time(); final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>(); for (List<IMetric> metrics: input) { final int batchId = batchIdGenerator.next(); final List<IMetric> batch = metrics; ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() { public Boolean call() throws Exception { final Timer.Context singleBatchWriteCtx = batchWriteDurationTimer.time(); try { metricsRWDelegator.insertMetrics(batch); final Timer.Context dirtyTimerCtx = slotUpdateTimer.time(); try { for (IMetric metric : batch) { context.update(metric.getCollectionTime(), Util.getShard(metric.getLocator().toString())); } } finally { dirtyTimerCtx.stop(); } return true; } catch (Exception ex) { getLogger().error(ex.getMessage(), ex); getLogger().warn("Did not persist all metrics successfully for batch " + batchId); if ( getLogger().isDebugEnabled() ) { List<String> failedLocators = new ArrayList<String>() {{ for (IMetric metric : batch) { add(metric.getLocator().toString()); } }}; getLogger().debug("Failed batch contains: " + Arrays.toString(failedLocators.toArray())); } return false; } finally { singleBatchWriteCtx.stop(); bufferedMetrics.dec(batch.size()); long now = System.currentTimeMillis(); if ( now - writeStartTime > timeout.toMillis()) { exceededScribeProcessingTime.mark(); getLogger().debug( String.format("Batch write time %d (ms) exceeded timeout %s before persisting " + "all metrics for batch %d", now - writeStartTime, timeout.toString(), batchId)); } } } }); resultFutures.add(futureBatchResult); } ListenableFuture<List<Boolean>> finalFuture = Futures.allAsList(resultFutures); finalFuture.addListener(new Runnable() { @Override public void run() { actualWriteCtx.stop(); } }, MoreExecutors.sameThreadExecutor()); return finalFuture; }
BatchWriter extends FunctionWithThreadPool<List<List<IMetric>>, ListenableFuture<List<Boolean>>> { @Override public ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input) throws Exception { final long writeStartTime = System.currentTimeMillis(); final Timer.Context actualWriteCtx = writeDurationTimer.time(); final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>(); for (List<IMetric> metrics: input) { final int batchId = batchIdGenerator.next(); final List<IMetric> batch = metrics; ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() { public Boolean call() throws Exception { final Timer.Context singleBatchWriteCtx = batchWriteDurationTimer.time(); try { metricsRWDelegator.insertMetrics(batch); final Timer.Context dirtyTimerCtx = slotUpdateTimer.time(); try { for (IMetric metric : batch) { context.update(metric.getCollectionTime(), Util.getShard(metric.getLocator().toString())); } } finally { dirtyTimerCtx.stop(); } return true; } catch (Exception ex) { getLogger().error(ex.getMessage(), ex); getLogger().warn("Did not persist all metrics successfully for batch " + batchId); if ( getLogger().isDebugEnabled() ) { List<String> failedLocators = new ArrayList<String>() {{ for (IMetric metric : batch) { add(metric.getLocator().toString()); } }}; getLogger().debug("Failed batch contains: " + Arrays.toString(failedLocators.toArray())); } return false; } finally { singleBatchWriteCtx.stop(); bufferedMetrics.dec(batch.size()); long now = System.currentTimeMillis(); if ( now - writeStartTime > timeout.toMillis()) { exceededScribeProcessingTime.mark(); getLogger().debug( String.format("Batch write time %d (ms) exceeded timeout %s before persisting " + "all metrics for batch %d", now - writeStartTime, timeout.toString(), batchId)); } } } }); resultFutures.add(futureBatchResult); } ListenableFuture<List<Boolean>> finalFuture = Futures.allAsList(resultFutures); finalFuture.addListener(new Runnable() { @Override public void run() { actualWriteCtx.stop(); } }, MoreExecutors.sameThreadExecutor()); return finalFuture; } }
BatchWriter extends FunctionWithThreadPool<List<List<IMetric>>, ListenableFuture<List<Boolean>>> { @Override public ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input) throws Exception { final long writeStartTime = System.currentTimeMillis(); final Timer.Context actualWriteCtx = writeDurationTimer.time(); final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>(); for (List<IMetric> metrics: input) { final int batchId = batchIdGenerator.next(); final List<IMetric> batch = metrics; ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() { public Boolean call() throws Exception { final Timer.Context singleBatchWriteCtx = batchWriteDurationTimer.time(); try { metricsRWDelegator.insertMetrics(batch); final Timer.Context dirtyTimerCtx = slotUpdateTimer.time(); try { for (IMetric metric : batch) { context.update(metric.getCollectionTime(), Util.getShard(metric.getLocator().toString())); } } finally { dirtyTimerCtx.stop(); } return true; } catch (Exception ex) { getLogger().error(ex.getMessage(), ex); getLogger().warn("Did not persist all metrics successfully for batch " + batchId); if ( getLogger().isDebugEnabled() ) { List<String> failedLocators = new ArrayList<String>() {{ for (IMetric metric : batch) { add(metric.getLocator().toString()); } }}; getLogger().debug("Failed batch contains: " + Arrays.toString(failedLocators.toArray())); } return false; } finally { singleBatchWriteCtx.stop(); bufferedMetrics.dec(batch.size()); long now = System.currentTimeMillis(); if ( now - writeStartTime > timeout.toMillis()) { exceededScribeProcessingTime.mark(); getLogger().debug( String.format("Batch write time %d (ms) exceeded timeout %s before persisting " + "all metrics for batch %d", now - writeStartTime, timeout.toString(), batchId)); } } } }); resultFutures.add(futureBatchResult); } ListenableFuture<List<Boolean>> finalFuture = Futures.allAsList(resultFutures); finalFuture.addListener(new Runnable() { @Override public void run() { actualWriteCtx.stop(); } }, MoreExecutors.sameThreadExecutor()); return finalFuture; } BatchWriter(ThreadPoolExecutor threadPool, TimeValue timeout, Counter bufferedMetrics, IngestionContext context); @VisibleForTesting  BatchWriter(ThreadPoolExecutor threadPool,
                       TimeValue timeout, Counter bufferedMetrics,
                       IngestionContext context,
                       MetricsRWDelegator metricsRWDelegator); }
BatchWriter extends FunctionWithThreadPool<List<List<IMetric>>, ListenableFuture<List<Boolean>>> { @Override public ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input) throws Exception { final long writeStartTime = System.currentTimeMillis(); final Timer.Context actualWriteCtx = writeDurationTimer.time(); final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>(); for (List<IMetric> metrics: input) { final int batchId = batchIdGenerator.next(); final List<IMetric> batch = metrics; ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() { public Boolean call() throws Exception { final Timer.Context singleBatchWriteCtx = batchWriteDurationTimer.time(); try { metricsRWDelegator.insertMetrics(batch); final Timer.Context dirtyTimerCtx = slotUpdateTimer.time(); try { for (IMetric metric : batch) { context.update(metric.getCollectionTime(), Util.getShard(metric.getLocator().toString())); } } finally { dirtyTimerCtx.stop(); } return true; } catch (Exception ex) { getLogger().error(ex.getMessage(), ex); getLogger().warn("Did not persist all metrics successfully for batch " + batchId); if ( getLogger().isDebugEnabled() ) { List<String> failedLocators = new ArrayList<String>() {{ for (IMetric metric : batch) { add(metric.getLocator().toString()); } }}; getLogger().debug("Failed batch contains: " + Arrays.toString(failedLocators.toArray())); } return false; } finally { singleBatchWriteCtx.stop(); bufferedMetrics.dec(batch.size()); long now = System.currentTimeMillis(); if ( now - writeStartTime > timeout.toMillis()) { exceededScribeProcessingTime.mark(); getLogger().debug( String.format("Batch write time %d (ms) exceeded timeout %s before persisting " + "all metrics for batch %d", now - writeStartTime, timeout.toString(), batchId)); } } } }); resultFutures.add(futureBatchResult); } ListenableFuture<List<Boolean>> finalFuture = Futures.allAsList(resultFutures); finalFuture.addListener(new Runnable() { @Override public void run() { actualWriteCtx.stop(); } }, MoreExecutors.sameThreadExecutor()); return finalFuture; } BatchWriter(ThreadPoolExecutor threadPool, TimeValue timeout, Counter bufferedMetrics, IngestionContext context); @VisibleForTesting  BatchWriter(ThreadPoolExecutor threadPool,
                       TimeValue timeout, Counter bufferedMetrics,
                       IngestionContext context,
                       MetricsRWDelegator metricsRWDelegator); @Override ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input); }
BatchWriter extends FunctionWithThreadPool<List<List<IMetric>>, ListenableFuture<List<Boolean>>> { @Override public ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input) throws Exception { final long writeStartTime = System.currentTimeMillis(); final Timer.Context actualWriteCtx = writeDurationTimer.time(); final List<ListenableFuture<Boolean>> resultFutures = new ArrayList<ListenableFuture<Boolean>>(); for (List<IMetric> metrics: input) { final int batchId = batchIdGenerator.next(); final List<IMetric> batch = metrics; ListenableFuture<Boolean> futureBatchResult = getThreadPool().submit(new Callable<Boolean>() { public Boolean call() throws Exception { final Timer.Context singleBatchWriteCtx = batchWriteDurationTimer.time(); try { metricsRWDelegator.insertMetrics(batch); final Timer.Context dirtyTimerCtx = slotUpdateTimer.time(); try { for (IMetric metric : batch) { context.update(metric.getCollectionTime(), Util.getShard(metric.getLocator().toString())); } } finally { dirtyTimerCtx.stop(); } return true; } catch (Exception ex) { getLogger().error(ex.getMessage(), ex); getLogger().warn("Did not persist all metrics successfully for batch " + batchId); if ( getLogger().isDebugEnabled() ) { List<String> failedLocators = new ArrayList<String>() {{ for (IMetric metric : batch) { add(metric.getLocator().toString()); } }}; getLogger().debug("Failed batch contains: " + Arrays.toString(failedLocators.toArray())); } return false; } finally { singleBatchWriteCtx.stop(); bufferedMetrics.dec(batch.size()); long now = System.currentTimeMillis(); if ( now - writeStartTime > timeout.toMillis()) { exceededScribeProcessingTime.mark(); getLogger().debug( String.format("Batch write time %d (ms) exceeded timeout %s before persisting " + "all metrics for batch %d", now - writeStartTime, timeout.toString(), batchId)); } } } }); resultFutures.add(futureBatchResult); } ListenableFuture<List<Boolean>> finalFuture = Futures.allAsList(resultFutures); finalFuture.addListener(new Runnable() { @Override public void run() { actualWriteCtx.stop(); } }, MoreExecutors.sameThreadExecutor()); return finalFuture; } BatchWriter(ThreadPoolExecutor threadPool, TimeValue timeout, Counter bufferedMetrics, IngestionContext context); @VisibleForTesting  BatchWriter(ThreadPoolExecutor threadPool,
                       TimeValue timeout, Counter bufferedMetrics,
                       IngestionContext context,
                       MetricsRWDelegator metricsRWDelegator); @Override ListenableFuture<List<Boolean>> apply(List<List<IMetric>> input);  }