@SuppressWarnings("unchecked") @Test public void uploadAssociationInfoTest() { PowerMockito.mockStatic(ConfigManager.class); List<String> types = new ArrayList<>(); types.add("type1"); when(ConfigManager.getTypes(anyString())).thenReturn(new HashSet<>(types)); when(ConfigManager.getKeyForType(anyString(), anyString())).thenReturn("type"); PowerMockito.mockStatic(ESManager.class); doNothing().when(ESManager.class); ESManager.createType(anyString(), anyString(),anyList()); List<Map<String, String>> entities = new ArrayList<>(); entities.add(new HashMap<>()); ESManager.uploadData(anyString(), anyString(), anyList(), anyString(), anyBoolean()); ESManager.deleteOldDocuments(anyString(), anyString(), anyString(), anyString()); new EntityAssociationManager().uploadAssociationInfo("dataSource","type"); }
public List<Map<String, String>> uploadAssociationInfo(String dataSource,String type) { LOGGER.info("Started EntityAssociationDataCollector for {}",type); List<Map<String,String>> errorList = new ArrayList<>(); AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials( new AWSStaticCredentialsProvider(new CredentialProvider().getCredentials(s3Account,s3Role))).withRegion(s3Region).build(); ObjectMapper objectMapper = new ObjectMapper(); try { String indexName = dataSource + "_" + type; String filePrefix = dataSource+"-"+type+"-"; List<String> childTypes = new ArrayList<>(); for (S3ObjectSummary objectSummary : s3Client.listObjectsV2(new ListObjectsV2Request().withBucketName(bucketName).withPrefix(dataPath+"/"+filePrefix)).getObjectSummaries()) { String fileName = objectSummary.getKey().replace(dataPath+"/", "").replace(".data", ""); if(fileName.chars().filter(ch -> ch == '-').count() == 2) { childTypes.add(fileName.replace(filePrefix,"")); } } String key = ConfigManager.getKeyForType(dataSource, type); if (!childTypes.isEmpty()) { for (String childType : childTypes) { String childTypeES = type+"_"+childType; if (!childType.equalsIgnoreCase("tags")) { ESManager.createType(indexName, childTypeES, type); LOGGER.info("Fetching data for {}", childTypeES); List<Map<String, Object>> entities = new ArrayList<>(); S3Object entitiesData = s3Client.getObject(new GetObjectRequest(bucketName, dataPath+"/"+filePrefix+childType+".data")); try (BufferedReader reader = new BufferedReader(new InputStreamReader(entitiesData.getObjectContent()))) { entities = objectMapper.readValue(reader.lines().collect(Collectors.joining("\n")),new TypeReference<List<Map<String, Object>>>() {}); } String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date()); entities.parallelStream().forEach(obj -> obj.put("_loaddate", loaddate)); LOGGER.info("Collected : {}", entities.size()); if (!entities.isEmpty()) { ErrorManager.getInstance(dataSource).handleError(indexName, childTypeES, loaddate, errorList,false); ESManager.uploadData(indexName, childTypeES, entities, key.split(",")); ESManager.deleteOldDocuments(indexName, childTypeES, "_loaddate.keyword", loaddate); } } } } } catch (Exception e) { LOGGER.error("Error in populating child tables", e); Map<String,String> errorMap = new HashMap<>(); errorMap.put(ERROR, "Error in populating child tables"); errorMap.put(ERROR_TYPE, WARN); errorMap.put(EXCEPTION, e.getMessage()); errorList.add(errorMap); } LOGGER.info("Completed EntityAssociationDataCollector for {}",type); return errorList; }
EntityAssociationManager implements Constants { public List<Map<String, String>> uploadAssociationInfo(String dataSource,String type) { LOGGER.info("Started EntityAssociationDataCollector for {}",type); List<Map<String,String>> errorList = new ArrayList<>(); AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials( new AWSStaticCredentialsProvider(new CredentialProvider().getCredentials(s3Account,s3Role))).withRegion(s3Region).build(); ObjectMapper objectMapper = new ObjectMapper(); try { String indexName = dataSource + "_" + type; String filePrefix = dataSource+"-"+type+"-"; List<String> childTypes = new ArrayList<>(); for (S3ObjectSummary objectSummary : s3Client.listObjectsV2(new ListObjectsV2Request().withBucketName(bucketName).withPrefix(dataPath+"/"+filePrefix)).getObjectSummaries()) { String fileName = objectSummary.getKey().replace(dataPath+"/", "").replace(".data", ""); if(fileName.chars().filter(ch -> ch == '-').count() == 2) { childTypes.add(fileName.replace(filePrefix,"")); } } String key = ConfigManager.getKeyForType(dataSource, type); if (!childTypes.isEmpty()) { for (String childType : childTypes) { String childTypeES = type+"_"+childType; if (!childType.equalsIgnoreCase("tags")) { ESManager.createType(indexName, childTypeES, type); LOGGER.info("Fetching data for {}", childTypeES); List<Map<String, Object>> entities = new ArrayList<>(); S3Object entitiesData = s3Client.getObject(new GetObjectRequest(bucketName, dataPath+"/"+filePrefix+childType+".data")); try (BufferedReader reader = new BufferedReader(new InputStreamReader(entitiesData.getObjectContent()))) { entities = objectMapper.readValue(reader.lines().collect(Collectors.joining("\n")),new TypeReference<List<Map<String, Object>>>() {}); } String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date()); entities.parallelStream().forEach(obj -> obj.put("_loaddate", loaddate)); LOGGER.info("Collected : {}", entities.size()); if (!entities.isEmpty()) { ErrorManager.getInstance(dataSource).handleError(indexName, childTypeES, loaddate, errorList,false); ESManager.uploadData(indexName, childTypeES, entities, key.split(",")); ESManager.deleteOldDocuments(indexName, childTypeES, "_loaddate.keyword", loaddate); } } } } } catch (Exception e) { LOGGER.error("Error in populating child tables", e); Map<String,String> errorMap = new HashMap<>(); errorMap.put(ERROR, "Error in populating child tables"); errorMap.put(ERROR_TYPE, WARN); errorMap.put(EXCEPTION, e.getMessage()); errorList.add(errorMap); } LOGGER.info("Completed EntityAssociationDataCollector for {}",type); return errorList; } }
EntityAssociationManager implements Constants { public List<Map<String, String>> uploadAssociationInfo(String dataSource,String type) { LOGGER.info("Started EntityAssociationDataCollector for {}",type); List<Map<String,String>> errorList = new ArrayList<>(); AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials( new AWSStaticCredentialsProvider(new CredentialProvider().getCredentials(s3Account,s3Role))).withRegion(s3Region).build(); ObjectMapper objectMapper = new ObjectMapper(); try { String indexName = dataSource + "_" + type; String filePrefix = dataSource+"-"+type+"-"; List<String> childTypes = new ArrayList<>(); for (S3ObjectSummary objectSummary : s3Client.listObjectsV2(new ListObjectsV2Request().withBucketName(bucketName).withPrefix(dataPath+"/"+filePrefix)).getObjectSummaries()) { String fileName = objectSummary.getKey().replace(dataPath+"/", "").replace(".data", ""); if(fileName.chars().filter(ch -> ch == '-').count() == 2) { childTypes.add(fileName.replace(filePrefix,"")); } } String key = ConfigManager.getKeyForType(dataSource, type); if (!childTypes.isEmpty()) { for (String childType : childTypes) { String childTypeES = type+"_"+childType; if (!childType.equalsIgnoreCase("tags")) { ESManager.createType(indexName, childTypeES, type); LOGGER.info("Fetching data for {}", childTypeES); List<Map<String, Object>> entities = new ArrayList<>(); S3Object entitiesData = s3Client.getObject(new GetObjectRequest(bucketName, dataPath+"/"+filePrefix+childType+".data")); try (BufferedReader reader = new BufferedReader(new InputStreamReader(entitiesData.getObjectContent()))) { entities = objectMapper.readValue(reader.lines().collect(Collectors.joining("\n")),new TypeReference<List<Map<String, Object>>>() {}); } String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date()); entities.parallelStream().forEach(obj -> obj.put("_loaddate", loaddate)); LOGGER.info("Collected : {}", entities.size()); if (!entities.isEmpty()) { ErrorManager.getInstance(dataSource).handleError(indexName, childTypeES, loaddate, errorList,false); ESManager.uploadData(indexName, childTypeES, entities, key.split(",")); ESManager.deleteOldDocuments(indexName, childTypeES, "_loaddate.keyword", loaddate); } } } } } catch (Exception e) { LOGGER.error("Error in populating child tables", e); Map<String,String> errorMap = new HashMap<>(); errorMap.put(ERROR, "Error in populating child tables"); errorMap.put(ERROR_TYPE, WARN); errorMap.put(EXCEPTION, e.getMessage()); errorList.add(errorMap); } LOGGER.info("Completed EntityAssociationDataCollector for {}",type); return errorList; }  }
EntityAssociationManager implements Constants { public List<Map<String, String>> uploadAssociationInfo(String dataSource,String type) { LOGGER.info("Started EntityAssociationDataCollector for {}",type); List<Map<String,String>> errorList = new ArrayList<>(); AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials( new AWSStaticCredentialsProvider(new CredentialProvider().getCredentials(s3Account,s3Role))).withRegion(s3Region).build(); ObjectMapper objectMapper = new ObjectMapper(); try { String indexName = dataSource + "_" + type; String filePrefix = dataSource+"-"+type+"-"; List<String> childTypes = new ArrayList<>(); for (S3ObjectSummary objectSummary : s3Client.listObjectsV2(new ListObjectsV2Request().withBucketName(bucketName).withPrefix(dataPath+"/"+filePrefix)).getObjectSummaries()) { String fileName = objectSummary.getKey().replace(dataPath+"/", "").replace(".data", ""); if(fileName.chars().filter(ch -> ch == '-').count() == 2) { childTypes.add(fileName.replace(filePrefix,"")); } } String key = ConfigManager.getKeyForType(dataSource, type); if (!childTypes.isEmpty()) { for (String childType : childTypes) { String childTypeES = type+"_"+childType; if (!childType.equalsIgnoreCase("tags")) { ESManager.createType(indexName, childTypeES, type); LOGGER.info("Fetching data for {}", childTypeES); List<Map<String, Object>> entities = new ArrayList<>(); S3Object entitiesData = s3Client.getObject(new GetObjectRequest(bucketName, dataPath+"/"+filePrefix+childType+".data")); try (BufferedReader reader = new BufferedReader(new InputStreamReader(entitiesData.getObjectContent()))) { entities = objectMapper.readValue(reader.lines().collect(Collectors.joining("\n")),new TypeReference<List<Map<String, Object>>>() {}); } String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date()); entities.parallelStream().forEach(obj -> obj.put("_loaddate", loaddate)); LOGGER.info("Collected : {}", entities.size()); if (!entities.isEmpty()) { ErrorManager.getInstance(dataSource).handleError(indexName, childTypeES, loaddate, errorList,false); ESManager.uploadData(indexName, childTypeES, entities, key.split(",")); ESManager.deleteOldDocuments(indexName, childTypeES, "_loaddate.keyword", loaddate); } } } } } catch (Exception e) { LOGGER.error("Error in populating child tables", e); Map<String,String> errorMap = new HashMap<>(); errorMap.put(ERROR, "Error in populating child tables"); errorMap.put(ERROR_TYPE, WARN); errorMap.put(EXCEPTION, e.getMessage()); errorList.add(errorMap); } LOGGER.info("Completed EntityAssociationDataCollector for {}",type); return errorList; }  List<Map<String, String>> uploadAssociationInfo(String dataSource,String type); }
EntityAssociationManager implements Constants { public List<Map<String, String>> uploadAssociationInfo(String dataSource,String type) { LOGGER.info("Started EntityAssociationDataCollector for {}",type); List<Map<String,String>> errorList = new ArrayList<>(); AmazonS3 s3Client = AmazonS3ClientBuilder.standard().withCredentials( new AWSStaticCredentialsProvider(new CredentialProvider().getCredentials(s3Account,s3Role))).withRegion(s3Region).build(); ObjectMapper objectMapper = new ObjectMapper(); try { String indexName = dataSource + "_" + type; String filePrefix = dataSource+"-"+type+"-"; List<String> childTypes = new ArrayList<>(); for (S3ObjectSummary objectSummary : s3Client.listObjectsV2(new ListObjectsV2Request().withBucketName(bucketName).withPrefix(dataPath+"/"+filePrefix)).getObjectSummaries()) { String fileName = objectSummary.getKey().replace(dataPath+"/", "").replace(".data", ""); if(fileName.chars().filter(ch -> ch == '-').count() == 2) { childTypes.add(fileName.replace(filePrefix,"")); } } String key = ConfigManager.getKeyForType(dataSource, type); if (!childTypes.isEmpty()) { for (String childType : childTypes) { String childTypeES = type+"_"+childType; if (!childType.equalsIgnoreCase("tags")) { ESManager.createType(indexName, childTypeES, type); LOGGER.info("Fetching data for {}", childTypeES); List<Map<String, Object>> entities = new ArrayList<>(); S3Object entitiesData = s3Client.getObject(new GetObjectRequest(bucketName, dataPath+"/"+filePrefix+childType+".data")); try (BufferedReader reader = new BufferedReader(new InputStreamReader(entitiesData.getObjectContent()))) { entities = objectMapper.readValue(reader.lines().collect(Collectors.joining("\n")),new TypeReference<List<Map<String, Object>>>() {}); } String loaddate = new SimpleDateFormat("yyyy-MM-dd HH:mm:00Z").format(new java.util.Date()); entities.parallelStream().forEach(obj -> obj.put("_loaddate", loaddate)); LOGGER.info("Collected : {}", entities.size()); if (!entities.isEmpty()) { ErrorManager.getInstance(dataSource).handleError(indexName, childTypeES, loaddate, errorList,false); ESManager.uploadData(indexName, childTypeES, entities, key.split(",")); ESManager.deleteOldDocuments(indexName, childTypeES, "_loaddate.keyword", loaddate); } } } } } catch (Exception e) { LOGGER.error("Error in populating child tables", e); Map<String,String> errorMap = new HashMap<>(); errorMap.put(ERROR, "Error in populating child tables"); errorMap.put(ERROR_TYPE, WARN); errorMap.put(EXCEPTION, e.getMessage()); errorList.add(errorMap); } LOGGER.info("Completed EntityAssociationDataCollector for {}",type); return errorList; }  List<Map<String, String>> uploadAssociationInfo(String dataSource,String type);  }