@SuppressWarnings("unchecked") @Test public void uploadEntityDataTest(){ PowerMockito.mockStatic(ConfigManager.class); List<String> types = new ArrayList<>(); types.add("onpremserver"); when(ConfigManager.getTypes(anyString())).thenReturn(new HashSet<>(types)); PowerMockito.mockStatic(ESManager.class); Map<String, Map<String, String>> currentInfo = new HashMap<>(); currentInfo.put("id",new HashMap<>()); when(ESManager.getExistingInfo(anyString(),anyString(), anyList())).thenReturn(currentInfo); List<Map<String, String>> tags = new ArrayList<>(); Map<String,String> tag = new HashMap<>(); tag.put("id", ""); tags.add(tag); PowerMockito.mockStatic(RDSDBManager.class); List<Map<String, String>> overridableInfo = new ArrayList<>(); Map<String,String> overridefields = new HashMap<>(); overridefields.put("updatableFields", "id,pac_override_id"); overridableInfo.add(overridefields); when(RDSDBManager.executeQuery(anyString())).thenReturn(overridableInfo); List<Map<String, String>> overrides = new ArrayList<>(); Map<String, String> override = new HashMap<>(); override.put("_resourceid", "id"); overrides.add(override); when(ConfigManager.getKeyForType(anyString(),anyString())).thenReturn("id"); when(ConfigManager.getIdForType(anyString(),anyString())).thenReturn("id"); new EntityManager().uploadEntityData("onpremserver"); }
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; }
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; } }
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; }  }
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; }  List<Map<String, String>> uploadEntityData(String datasource); }
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; }  List<Map<String, String>> uploadEntityData(String datasource);  }