class EntityManager implements Constants {

    public List<Map<String, String>> uploadEntityData(String datasource) {
    	List<Map<String,String>> errorList = new ArrayList<>();
        Set<String> types = ConfigManager.getTypes(datasource);
        Iterator<String> itr = types.iterator();
        String type = "";
        LOGGER.info("*** Start Colleting Entity Info ***");
        List<String> filters = Arrays.asList("_docid", FIRST_DISCOVERED);
        EntityAssociationManager childTypeManager = new EntityAssociationManager();
        while (itr.hasNext()) {
            try {
                type = itr.next();
                Map<String, Object> stats = new LinkedHashMap<>();
            	String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date());
                stats.put("datasource", datasource);
                stats.put("type", type);
                stats.put("start_time",  new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ").format(new java.util.Date()));
                LOGGER.info("Fetching {}" , type);
                String indexName = datasource + "_" + type;
                Map<String, Map<String, String>> currentInfo = ESManager.getExistingInfo(indexName, type, filters);
                LOGGER.info("Existing no of docs : {}" , currentInfo.size());
                
                List<Map<String, Object>> entities = fetchEntitiyInfoFromS3(datasource,type,errorList);
                List<Map<String, String>> tags = fetchTagsForEntitiesFromS3(datasource, type);
                
                LOGGER.info("Fetched from S3");
                if(!entities.isEmpty()){
	                List<Map<String, String>> overridableInfo = RDSDBManager.executeQuery(
	                        "select updatableFields  from cf_pac_updatable_fields where resourceType ='" + type + "'");
	                List<Map<String, String>> overrides = RDSDBManager.executeQuery(
	                        "select _resourceid,fieldname,fieldvalue from pacman_field_override where resourcetype = '"
	                                + type + "'");
	                Map<String, List<Map<String, String>>> overridesMap = overrides.parallelStream()
	                        .collect(Collectors.groupingBy(obj -> obj.get("_resourceid")));
	                
	                String keys = ConfigManager.getKeyForType(datasource, type); 
	                String idColumn = ConfigManager.getIdForType(datasource, type);
	                String[] keysArray = keys.split(",");
	                
	                prepareDocs(currentInfo, entities, tags, overridableInfo, overridesMap, idColumn, keysArray, type);
	                Map<String,Long> errUpdateInfo = ErrorManager.getInstance(datasource).handleError(indexName,type,loaddate,errorList,true);
	                Map<String, Object> uploadInfo = ESManager.uploadData(indexName, type, entities, loaddate);
	                stats.putAll(uploadInfo);
	                stats.put("errorUpdates", errUpdateInfo);
	                errorList.addAll(childTypeManager.uploadAssociationInfo(datasource, type)) ;
	                
                } 
                stats.put("total_docs", entities.size());
                stats.put("end_time", new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ").format(new java.util.Date()));
                stats.put("newly_discovered",entities.stream().filter(entity->entity.get(DISCOVERY_DATE).equals(entity.get(FIRST_DISCOVERED))).count());
                String statsJson = ESManager.createESDoc(stats);
                ESManager.invokeAPI("POST", "/datashipper/stats", statsJson);
            } catch (Exception e) {
                LOGGER.error("Exception in collecting/uploading data for {}" ,type,e);
                Map<String,String> errorMap = new HashMap<>();
                errorMap.put(ERROR, "Exception in collecting/uploading data for "+type);
                errorMap.put(ERROR_TYPE, WARN);
                errorMap.put(EXCEPTION, e.getMessage());
                errorList.add(errorMap);
            }
           
        }
        LOGGER.info("*** End Colleting Entity Info ***");
        return errorList;
    }

}