Merge pull request #283 from amitpriyadarshi/release-3.0.0

Merge branch 'release-2.10.0' into release-3.0.0

ontology-engine/graph-dac-api/src/main/java/org/sunbird/graph/dac/model/Relation.java

 package org.sunbird.graph.dac.model;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.neo4j.driver.v1.Value;
 import org.neo4j.graphdb.Node;
@@ -230,8 +231,18 @@ public class Relation implements Serializable {
 		this.endNodeId = endNodeId;
 	}
 
+	// TODO: In 3.0 if metadata is empty set it with new HashMap and return (to handle NPE.
+//	public Map<String, Object> getMetadata() {
+//		if (MapUtils.isEmpty(metadata))
+//			metadata = new HashMap<String, Object>();
+//		return metadata;
+//	}
+
 	public Map<String, Object> getMetadata() {
-		return metadata;
+		if (!MapUtils.isEmpty(metadata))
+			return metadata;
+		else
+			return new HashMap<String, Object>();
 	}
 
 	public void setMetadata(Map<String, Object> metadata) {

search-api/search-actors/src/main/java/org/sunbird/actors/SearchActor.java

@@ -2,6 +2,7 @@ package org.sunbird.actors;
 
 import akka.dispatch.Futures;
 import akka.dispatch.Mapper;
+import akka.dispatch.Recover;
 import akka.util.Timeout;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -36,43 +37,54 @@ public class SearchActor extends SearchBaseActor {
     public Future<Response> onReceive(Request request) throws Throwable {
         String operation = request.getOperation();
         SearchProcessor processor = new SearchProcessor();
-        if (StringUtils.equalsIgnoreCase("INDEX_SEARCH", operation)) {
-            SearchDTO searchDTO = getSearchDTO(request);
-            Future<Map<String, Object>> searchResult = processor.processSearch(searchDTO, true);
-            return searchResult.map(new Mapper<Map<String, Object>, Response>() {
-                @Override
-                public Response apply(Map<String, Object> lstResult) {
-                    String mode = (String) request.getRequest().get(SearchConstants.mode);
-                    if (StringUtils.isNotBlank(mode) && StringUtils.equalsIgnoreCase("collection", mode)) {
-                        return OK(getCollectionsResult(lstResult, processor, request));
-                    } else {
-                        return OK(lstResult);
+        try{
+            if (StringUtils.equalsIgnoreCase("INDEX_SEARCH", operation)) {
+                SearchDTO searchDTO = getSearchDTO(request);
+                Future<Map<String, Object>> searchResult = processor.processSearch(searchDTO, true);
+                return searchResult.map(new Mapper<Map<String, Object>, Response>() {
+                    @Override
+                    public Response apply(Map<String, Object> lstResult) {
+                        String mode = (String) request.getRequest().get(SearchConstants.mode);
+                        if (StringUtils.isNotBlank(mode) && StringUtils.equalsIgnoreCase("collection", mode)) {
+                            return OK(getCollectionsResult(lstResult, processor, request));
+                        } else {
+                            return OK(lstResult);
+                        }
                     }
+                }, getContext().dispatcher()).recoverWith(new Recover<Future<Response>>() {
+                    @Override
+                    public Future<Response> recover(Throwable failure) throws Throwable {
+                        TelemetryManager.error("Unable to process the request:: Request: " + JsonUtils.serialize(request), failure);
+                        return ERROR(request.getOperation(), failure);
+                    }
+                }, getContext().dispatcher());
+            } else if (StringUtils.equalsIgnoreCase("COUNT", operation)) {
+                Map<String, Object> countResult = processor.processCount(getSearchDTO(request));
+                if (null != countResult.get("count")) {
+                    Integer count = (Integer) countResult.get("count");
+                    return Futures.successful(OK("count", count));
+                } else {
+                    return Futures.successful(ERROR("", "count is empty or null", ResponseCode.SERVER_ERROR, "", null));
                 }
-            }, getContext().dispatcher());
-        } else if (StringUtils.equalsIgnoreCase("COUNT", operation)) {
-            Map<String, Object> countResult = processor.processCount(getSearchDTO(request));
-            if (null != countResult.get("count")) {
-                Integer count = (Integer) countResult.get("count");
-                return Futures.successful(OK("count", count));
+            } else if (StringUtils.equalsIgnoreCase("METRICS", operation)) {
+                Future<Map<String, Object>> searchResult = processor.processSearch(getSearchDTO(request), false);
+                return searchResult.map(new Mapper<Map<String, Object>, Response>() {
+                    @Override
+                    public Response apply(Map<String, Object> lstResult) {
+                        return OK(getCompositeSearchResponse(lstResult));
+                    }
+                }, getContext().dispatcher());
+            } else if (StringUtils.equalsIgnoreCase("GROUP_SEARCH_RESULT_BY_OBJECTTYPE", operation)) {
+                Map<String, Object> searchResponse = (Map<String, Object>) request.get("searchResult");
+                return Futures.successful(OK(getCompositeSearchResponse(searchResponse)));
             } else {
-                return Futures.successful(ERROR("", "count is empty or null", ResponseCode.SERVER_ERROR, "", null));
+                TelemetryManager.log("Unsupported operation: " + operation);
+                throw new ClientException(SearchConstants.ERR_INVALID_OPERATION,
+                        "Unsupported operation: " + operation);
             }
-        } else if (StringUtils.equalsIgnoreCase("METRICS", operation)) {
-            Future<Map<String, Object>> searchResult = processor.processSearch(getSearchDTO(request), false);
-            return searchResult.map(new Mapper<Map<String, Object>, Response>() {
-                @Override
-                public Response apply(Map<String, Object> lstResult) {
-                    return OK(getCompositeSearchResponse(lstResult));
-                }
-            }, getContext().dispatcher());
-        } else if (StringUtils.equalsIgnoreCase("GROUP_SEARCH_RESULT_BY_OBJECTTYPE", operation)) {
-            Map<String, Object> searchResponse = (Map<String, Object>) request.get("searchResult");
-            return Futures.successful(OK(getCompositeSearchResponse(searchResponse)));
-        } else {
-            TelemetryManager.log("Unsupported operation: " + operation);
-            throw new ClientException(SearchConstants.ERR_INVALID_OPERATION,
-                    "Unsupported operation: " + operation);
+        } catch (Exception e) {
+            TelemetryManager.info("Error while processing the request: REQUEST::" + JsonUtils.serialize(request));
+            return ERROR(operation, e);
         }
     }
 
@@ -443,6 +455,7 @@ public class SearchActor extends SearchBaseActor {
                                     break;
                                 }
                                 default: {
+                                    TelemetryManager.error("Invalid filters, Unsupported operation:: " + filterEntry.getKey() + ":: filters::" + filters);
                                     throw new Exception("Unsupported operation");
                                 }
                             }

search-api/search-actors/src/main/java/org/sunbird/actors/SearchBaseActor.java

@@ -3,8 +3,9 @@ package org.sunbird.actors;
 import akka.actor.AbstractActor;
 import akka.actor.ActorRef;
 import akka.dispatch.Futures;
+import akka.dispatch.Recover;
 import akka.pattern.Patterns;
-import org.elasticsearch.action.search.SearchPhaseExecutionException;
+import org.sunbird.common.JsonUtils;
 import org.sunbird.common.dto.Request;
 import org.sunbird.common.dto.Response;
 import org.sunbird.common.dto.ResponseParams;
@@ -36,7 +37,13 @@ public abstract class SearchBaseActor extends AbstractActor {
 
     private Future<Response> internalOnReceive(Request request) {
         try {
-            return onReceive(request);
+            return onReceive(request).recoverWith(new Recover<Future<Response>>() {
+                @Override
+                public Future<Response> recover(Throwable failure) throws Throwable {
+                    TelemetryManager.error("Unable to process the request:: Request: " + JsonUtils.serialize(request), failure);
+                    return ERROR(request.getOperation(), failure);
+                }
+            }, getContext().dispatcher());
         } catch (Throwable e) {
             return ERROR(request.getOperation(), e);
         }
@@ -67,10 +74,10 @@ public abstract class SearchBaseActor extends AbstractActor {
             params.setErr(mwException.getErrCode());
             response.put("messages", mwException.getMessage());
         } else {
-            e.printStackTrace();
+            TelemetryManager.error("Error while processing", e);
             params.setErr("ERR_SYSTEM_EXCEPTION");
         }
-        System.out.println("Exception occurred - class :" + e.getClass().getName() + " with message :" + e.getMessage());
+        TelemetryManager.error("Exception occurred - class :" + e.getClass().getName() + " with message :" , e);
         params.setErrmsg(setErrMessage(e));
         response.setParams(params);
         setResponseCode(response, e);
@@ -152,9 +159,12 @@ public abstract class SearchBaseActor extends AbstractActor {
     protected String setErrMessage(Throwable e){
 		if (e instanceof MiddlewareException)
         		return e.getMessage();
-		else if  (e instanceof SearchPhaseExecutionException)
-			return e.getCause().getMessage();
-         else 
-        	 	return "Something went wrong in server while processing the request";
+        else {
+            if (e.getSuppressed().length > 0) {
+                return e.getSuppressed()[0].getMessage();
+            } else {
+                return e.getMessage();
+            }
+        }
     }
 }

search-api/search-core/src/main/java/org/sunbird/search/client/ElasticSearchUtil.java

@@ -9,7 +9,9 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpHost;
+import org.apache.http.client.config.RequestConfig;
 import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.RestClientBuilder;
 import org.sunbird.search.util.SearchConstants;
 import org.sunbird.telemetry.logger.TelemetryManager;
 import org.elasticsearch.action.ActionListener;
@@ -102,8 +104,14 @@ public class ElasticSearchUtil {
 			for (String host : hostPort.keySet()) {
 				httpHosts.add(new HttpHost(host, hostPort.get(host)));
 			}
-			RestHighLevelClient client = new RestHighLevelClient(
-					RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()])));
+			RestClientBuilder builder = RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]))
+					.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+						@Override
+						public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
+							return requestConfigBuilder.setConnectionRequestTimeout(-1);
+						}
+					});
+			RestHighLevelClient client = new RestHighLevelClient(builder);
 			if (null != client)
 				esClient.put(indexName, client);
 		}

search-api/search-core/src/main/java/org/sunbird/search/util/DefinitionUtil.java

@@ -24,7 +24,7 @@ public class DefinitionUtil {
         String version = objectTypeSchemaMap.get(objectType.toLowerCase()).getOrDefault("version", "1.0");
         ISchemaValidator schemaValidator = SchemaValidatorFactory.getInstance(schemaName, version);
         if(schemaValidator.getConfig().hasPath("searchProps")){
-            metadata.putAll(schemaValidator.getConfig().getObject("searchProps"));
+            metadata.putAll(schemaValidator.getConfig().getObject("searchProps").unwrapped());
         }
         return metadata;
     }