DefaultCollectionService extends DefaultJPAService implements CollectionService { @Override public int commitMetricSchema(int messageCount, int timeout) { requireNotDisposed(); requireArgument(messageCount > 0, "Message count must be greater than zero."); requireArgument(timeout > 0, "The timeout in milliseconds must be greater than zero."); CollectionType type = new ObjectMapper().getTypeFactory().constructCollectionType(ArrayList.class, Metric.class); List<ArrayList<Metric>> dequeuedMessages = _mqService.dequeue(METRIC.getQueueName(), type, timeout, messageCount); List<Metric> dequeued = new ArrayList<Metric>(); for (List<Metric> list : dequeuedMessages) { dequeued.addAll(list); } if (!dequeued.isEmpty()) { _schemaService.put(dequeued); _logger.debug("Committed {} metrics for schema records creation.", dequeued.size()); } return dequeued.size(); } @Inject  DefaultCollectionService(MQService mqService, TSDBService tsdbService, AuditService auditService,
            AnnotationStorageService annotationStorageService, SystemConfiguration configuration, SchemaService schemaService, WardenService wardenService,
            MonitorService monitorService, NamespaceService namespaceService); @Override void submitMetric(PrincipalUser submitter, Metric metric); @Override void submitMetrics(PrincipalUser submitter, List<Metric> metrics); @Override void submitAnnotation(PrincipalUser submitter, Annotation annotation); @Override void submitAnnotations(PrincipalUser submitter, List<Annotation> annotations); @Override List<Metric> commitMetrics(int messageCount, int timeout); @Override int commitMetricSchema(int messageCount, int timeout); @Override int commitAnnotations(int annotationCount, int timeout); @Override int commitHistograms(int histogramCount, int timeout); @Override void submitHistogram(PrincipalUser submitter, Histogram histogram); @Override void submitHistograms(PrincipalUser submitter, List<Histogram> histograms); @Override void dispose();  }
@Test public void testCommitMetricSchema() { List<Serializable> messages = Arrays.asList( new ArrayList<>(Arrays.asList(TestUtils.createMetric())), new ArrayList<>(Arrays.asList(TestUtils.createMetric())) ); when(mqService.dequeue(eq(MQService.MQQueue.METRIC.getQueueName()), any(CollectionType.class), anyInt(), anyInt())).thenReturn(messages); assertEquals(2, collectionService.commitMetricSchema(2, 60000)); }