code: protected List<List<MetricSchemaRecord>> _fracture(List<Metric> metrics) {
  		List<List<MetricSchemaRecord>> fracturedList = new ArrayList<>();
  
  		List<MetricSchemaRecord> records = new ArrayList<>(_bulkIndexingSize);
  		for(Metric metric : metrics) {
  			if(metric.getTags().isEmpty()) {
  				MetricSchemaRecord msr = new MetricSchemaRecord(metric.getScope(), metric.getMetric());
  				msr.setNamespace(metric.getNamespace());
  				records.add(msr);
  				if(records.size() == _bulkIndexingSize) {
  					fracturedList.add(records);
  					records = new ArrayList<>(_bulkIndexingSize);
  				}
  				continue;
  			}
  
  			String retention  = metric.getMetatagsRecord()==null?null:metric.getMetatagsRecord().getMetatagValue(RETENTION_DISCOVERY);
  			Integer retentionInt = null;
  			if (retention != null) {
  				try {
  					retentionInt = Integer.parseInt(retention);
  				}
  				catch(NumberFormatException e) {
  					_logger.debug("expect _retention_discovery_ to be a numeric value; {} is invalid", retention);
  				}
  			}
  			for(Map.Entry<String, String> entry : metric.getTags().entrySet()) {
  				records.add(new MetricSchemaRecord(metric.getNamespace(),
  													metric.getScope(),
  													metric.getMetric(),
  													entry.getKey(),
  													entry.getValue(),
  													retentionInt));
  				if(records.size() == _bulkIndexingSize) {
  					fracturedList.add(records);
  					records = new ArrayList<>(_bulkIndexingSize);
  				}
  			}
  		}
  
  		if(!records.isEmpty()) {
  			fracturedList.add(records);
  		}
  
  		return fracturedList;
  	}
docs: /* Convert the given list of metrics to a list of metric schema records. At the same time, fracture the records list
  if its size is greater than ELASTICSEARCH_INDEXING_BATCH_SIZE.