@SuppressWarnings({ "unchecked", "static-access" }) @Test public void getExistingInfoTest() throws Exception{ HttpEntity jsonEntity = new StringEntity("{}", ContentType.APPLICATION_JSON); when(response.getEntity()).thenReturn(jsonEntity); when(restClient.performRequest(anyString(), anyString(), anyMap(), any(HttpEntity.class), Matchers.<Header>anyVararg())).thenReturn(response); ReflectionTestUtils.setField(esManager, "restClient", restClient); when(sl.getStatusCode()).thenReturn(100); when(response.getStatusLine()).thenReturn(sl); String countString = "{\"count\":\"1\"}"; String jsonString = "{\"hits\":{\"hits\":[{\"k1\":\"v1\",\"k2\":\"v2\"}]}}"; JsonNode countObj = new ObjectMapper().readTree(countString); JsonNode jsonObj = new ObjectMapper().readTree(jsonString); when(mapper.readTree(anyString())).thenReturn(countObj,jsonObj); List<String> filters = new ArrayList<>(); filters.add("filter"); esManager.getExistingInfo("indexName", "type", filters); }
public static Map<String, Map<String, String>> getExistingInfo(String indexName, String type, List<String> filters) { int count = getTypeCount(indexName, type); int _count = count; boolean scroll = false; if (count > 10000) { _count = 10000; scroll = true; } String keyField = filters.get(0); StringBuilder filter_path = new StringBuilder("&filter_path=_scroll_id,"); for (String _filter : filters) { filter_path.append("hits.hits._source." + _filter + ","); } filter_path.deleteCharAt(filter_path.length() - 1); String endPoint = indexName + "/" + type + "/_search?scroll=1m" + filter_path.toString() + "&size=" + _count; String payLoad = "{ \"query\": { \"match\": {\"latest\": true}}}"; Map<String, Map<String, String>> _data = new HashMap<>(); String scrollId = fetchDataAndScrollId(endPoint, _data, keyField, payLoad); if (scroll) { count -= 10000; do { endPoint = "/_search/scroll?scroll=1m&scroll_id=" + scrollId + filter_path.toString(); scrollId = fetchDataAndScrollId(endPoint, _data, keyField, null); count -= 10000; if (count <= 0) scroll = false; } while (scroll); } return _data; }
ESManager implements Constants { public static Map<String, Map<String, String>> getExistingInfo(String indexName, String type, List<String> filters) { int count = getTypeCount(indexName, type); int _count = count; boolean scroll = false; if (count > 10000) { _count = 10000; scroll = true; } String keyField = filters.get(0); StringBuilder filter_path = new StringBuilder("&filter_path=_scroll_id,"); for (String _filter : filters) { filter_path.append("hits.hits._source." + _filter + ","); } filter_path.deleteCharAt(filter_path.length() - 1); String endPoint = indexName + "/" + type + "/_search?scroll=1m" + filter_path.toString() + "&size=" + _count; String payLoad = "{ \"query\": { \"match\": {\"latest\": true}}}"; Map<String, Map<String, String>> _data = new HashMap<>(); String scrollId = fetchDataAndScrollId(endPoint, _data, keyField, payLoad); if (scroll) { count -= 10000; do { endPoint = "/_search/scroll?scroll=1m&scroll_id=" + scrollId + filter_path.toString(); scrollId = fetchDataAndScrollId(endPoint, _data, keyField, null); count -= 10000; if (count <= 0) scroll = false; } while (scroll); } return _data; } }
ESManager implements Constants { public static Map<String, Map<String, String>> getExistingInfo(String indexName, String type, List<String> filters) { int count = getTypeCount(indexName, type); int _count = count; boolean scroll = false; if (count > 10000) { _count = 10000; scroll = true; } String keyField = filters.get(0); StringBuilder filter_path = new StringBuilder("&filter_path=_scroll_id,"); for (String _filter : filters) { filter_path.append("hits.hits._source." + _filter + ","); } filter_path.deleteCharAt(filter_path.length() - 1); String endPoint = indexName + "/" + type + "/_search?scroll=1m" + filter_path.toString() + "&size=" + _count; String payLoad = "{ \"query\": { \"match\": {\"latest\": true}}}"; Map<String, Map<String, String>> _data = new HashMap<>(); String scrollId = fetchDataAndScrollId(endPoint, _data, keyField, payLoad); if (scroll) { count -= 10000; do { endPoint = "/_search/scroll?scroll=1m&scroll_id=" + scrollId + filter_path.toString(); scrollId = fetchDataAndScrollId(endPoint, _data, keyField, null); count -= 10000; if (count <= 0) scroll = false; } while (scroll); } return _data; }  }
ESManager implements Constants { public static Map<String, Map<String, String>> getExistingInfo(String indexName, String type, List<String> filters) { int count = getTypeCount(indexName, type); int _count = count; boolean scroll = false; if (count > 10000) { _count = 10000; scroll = true; } String keyField = filters.get(0); StringBuilder filter_path = new StringBuilder("&filter_path=_scroll_id,"); for (String _filter : filters) { filter_path.append("hits.hits._source." + _filter + ","); } filter_path.deleteCharAt(filter_path.length() - 1); String endPoint = indexName + "/" + type + "/_search?scroll=1m" + filter_path.toString() + "&size=" + _count; String payLoad = "{ \"query\": { \"match\": {\"latest\": true}}}"; Map<String, Map<String, String>> _data = new HashMap<>(); String scrollId = fetchDataAndScrollId(endPoint, _data, keyField, payLoad); if (scroll) { count -= 10000; do { endPoint = "/_search/scroll?scroll=1m&scroll_id=" + scrollId + filter_path.toString(); scrollId = fetchDataAndScrollId(endPoint, _data, keyField, null); count -= 10000; if (count <= 0) scroll = false; } while (scroll); } return _data; }  static Map<String, Object> uploadData(String index, String type, List<Map<String, Object>> docs, String loaddate); static void refresh(String index); static void uploadData(String index, String type, List<Map<String, Object>> docs, String idKey,
            boolean refresh); static String createESDoc(Map<String, ?> doc); static Response invokeAPI(String method, String endpoint, String payLoad); static void configureIndexAndTypes(String ds, List<Map<String, String>> errorList); static Map<String, Map<String, String>> getExistingInfo(String indexName, String type,
            List<String> filters); @SuppressWarnings("unchecked") static Map<String, Map<String, Map<String, Object>>> fetchCurrentCountStatsForAssetGroups(String date); static void createIndex(String indexName, List<Map<String, String>> errorList); static void createType(String indexName, String typename, List<Map<String, String>> errorList); static void createType(String index, String type, String parent); static void uploadData(String index, String type, List<Map<String, Object>> docs, String[] parentKey); static void deleteOldDocuments(String index, String type, String field, String value); static long updateLoadDate(String index, String type, String accountId, String region, String loaddate,boolean checkLatest); }
ESManager implements Constants { public static Map<String, Map<String, String>> getExistingInfo(String indexName, String type, List<String> filters) { int count = getTypeCount(indexName, type); int _count = count; boolean scroll = false; if (count > 10000) { _count = 10000; scroll = true; } String keyField = filters.get(0); StringBuilder filter_path = new StringBuilder("&filter_path=_scroll_id,"); for (String _filter : filters) { filter_path.append("hits.hits._source." + _filter + ","); } filter_path.deleteCharAt(filter_path.length() - 1); String endPoint = indexName + "/" + type + "/_search?scroll=1m" + filter_path.toString() + "&size=" + _count; String payLoad = "{ \"query\": { \"match\": {\"latest\": true}}}"; Map<String, Map<String, String>> _data = new HashMap<>(); String scrollId = fetchDataAndScrollId(endPoint, _data, keyField, payLoad); if (scroll) { count -= 10000; do { endPoint = "/_search/scroll?scroll=1m&scroll_id=" + scrollId + filter_path.toString(); scrollId = fetchDataAndScrollId(endPoint, _data, keyField, null); count -= 10000; if (count <= 0) scroll = false; } while (scroll); } return _data; }  static Map<String, Object> uploadData(String index, String type, List<Map<String, Object>> docs, String loaddate); static void refresh(String index); static void uploadData(String index, String type, List<Map<String, Object>> docs, String idKey,
            boolean refresh); static String createESDoc(Map<String, ?> doc); static Response invokeAPI(String method, String endpoint, String payLoad); static void configureIndexAndTypes(String ds, List<Map<String, String>> errorList); static Map<String, Map<String, String>> getExistingInfo(String indexName, String type,
            List<String> filters); @SuppressWarnings("unchecked") static Map<String, Map<String, Map<String, Object>>> fetchCurrentCountStatsForAssetGroups(String date); static void createIndex(String indexName, List<Map<String, String>> errorList); static void createType(String indexName, String typename, List<Map<String, String>> errorList); static void createType(String index, String type, String parent); static void uploadData(String index, String type, List<Map<String, Object>> docs, String[] parentKey); static void deleteOldDocuments(String index, String type, String field, String value); static long updateLoadDate(String index, String type, String accountId, String region, String loaddate,boolean checkLatest);  }