Merge branch 'requeue_logic' into 'develop'

failed record sending back to record queue

See merge request osdu/core/os-indexer!9

provider/indexer-ibm/src/main/java/org/opengroup/osdu/indexer/ibm/util/GlobalExceptionMapper.java

+// Copyright 2017-2019, Schlumberger
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package org.opengroup.osdu.indexer.ibm.util;
+
+import javax.validation.ValidationException;
+
+import javassist.NotFoundException;
+import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.AccessDeniedException;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.servlet.mvc.method.annotation.ResponseEntityExceptionHandler;
+import org.opengroup.osdu.core.common.model.http.AppException;
+
+@Order(Ordered.HIGHEST_PRECEDENCE)
+@ControllerAdvice
+public class GlobalExceptionMapper extends ResponseEntityExceptionHandler {
+
+    @Autowired
+    private JaxRsDpsLog logger;
+
+    @ExceptionHandler(AppException.class)
+    protected ResponseEntity<Object> handleAppException(AppException e) {
+        return this.getErrorResponse(e);
+    }
+
+    @ExceptionHandler(ValidationException.class)
+    protected ResponseEntity<Object> handleValidationException(ValidationException e) {
+        return this.getErrorResponse(
+                new AppException(HttpStatus.BAD_REQUEST.value(), "Validation error.", e.getMessage(), e));
+    }
+
+    @ExceptionHandler(NotFoundException.class)
+    protected ResponseEntity<Object> handleNotFoundException(NotFoundException e) {
+        return this.getErrorResponse(
+                new AppException(HttpStatus.NOT_FOUND.value(), "Resource not found.", e.getMessage(), e));
+    }
+
+    @ExceptionHandler(AccessDeniedException.class)
+    protected ResponseEntity<Object> handleAccessDeniedException(AccessDeniedException e) {
+        return this.getErrorResponse(
+                new AppException(HttpStatus.FORBIDDEN.value(), "Access denied", e.getMessage(), e));
+    }
+
+    @ExceptionHandler(Exception.class)
+    protected ResponseEntity<Object> handleGeneralException(Exception e) {
+        return this.getErrorResponse(
+                new AppException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "Server error.",
+                        "An unknown error has occurred.", e));
+    }
+
+    private ResponseEntity<Object> getErrorResponse(AppException e) {
+
+        String exceptionMsg = e.getOriginalException() != null
+                ? e.getOriginalException().getMessage()
+                : e.getError().getMessage();
+
+        if (e.getError().getCode() > 499) {
+            this.logger.error(exceptionMsg, e);
+        } else {
+            this.logger.warning(exceptionMsg, e);
+        }
+
+        return new ResponseEntity<Object>(e.getError(), HttpStatus.resolve(e.getError().getCode()));
+    }
+}
\ No newline at end of file

provider/indexer-ibm/src/main/java/org/opengroup/osdu/indexer/ibm/util/IndexerQueueTaskBuilderIbm.java

+package org.opengroup.osdu.indexer.ibm.util;
+
+import java.util.Map;
+
+import javax.inject.Inject;
+
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
+import org.opengroup.osdu.core.ibm.messagebus.IMessageFactory;
+import org.opengroup.osdu.indexer.util.IndexerQueueTaskBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Component;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonSyntaxException;
+
+@Primary
+@Component
+public class IndexerQueueTaskBuilderIbm extends IndexerQueueTaskBuilder {
+	private static final Logger logger = LoggerFactory.getLogger(IndexerQueueTaskBuilderIbm.class);
+	
+	@Inject
+	IMessageFactory mq;
+	
+	private Gson gson;
+	private final static String RETRY_STRING = "retry";
+	private final static String ERROR_CODE = "errorCode";
+	private final static String ERROR_MESSAGE = "errorMessage";
+
+	@Inject
+	public void init() {
+		gson = new Gson();
+	}
+
+	@Override
+	public void createWorkerTask(String payload, DpsHeaders headers) {
+		createTask(payload, headers);
+	}
+
+	@Override
+	public void createReIndexTask(String payload, DpsHeaders headers) {
+		createTask(payload, headers);
+	}
+
+	private void createTask(String payload, DpsHeaders headers) {
+
+		try {
+			RecordChangedMessages receivedPayload = gson.fromJson(payload, RecordChangedMessages.class);
+
+			Map<String, String> attributes = receivedPayload.getAttributes();
+			int retryCount = 0;
+			if (attributes.containsKey(RETRY_STRING)) {
+				retryCount = Integer.parseInt(attributes.get(RETRY_STRING));
+				retryCount++;
+			} else {
+				retryCount = 1;
+			}
+			attributes.put(RETRY_STRING, String.valueOf(retryCount));
+			attributes.put(ERROR_CODE, "999"); //error code TBD 
+			attributes.put(ERROR_MESSAGE, "Indexer could not process record");
+			receivedPayload.setAttributes(attributes);
+
+			// incase if we need to shift logic from indexer-queue-ibm/subscriber.java
+			/*
+			 * if(Integer.parseInt(receivedPayload.getAttributes().get(RETRY_STRING))>3) {
+			 * //add DLQ in IMessageFactory
+			 * 
+			 * mq.sendMessage("DLQ", gson.toJson(receivedPayload)); }
+			 */
+			logger.info("Message send back to queue : " + receivedPayload);
+			mq.sendMessage(IMessageFactory.DEFAULT_QUEUE_NAME, gson.toJson(receivedPayload));
+		} catch (JsonSyntaxException e) {
+			logger.error("JsonSyntaxException in IndexerQueueTaskBuilderIbm " + e.toString());
+			e.printStackTrace();
+		} catch (NumberFormatException e) {
+			logger.error("NumberFormatException in IndexerQueueTaskBuilderIbm " + e.toString());
+			e.printStackTrace();
+		} catch (Exception e) {
+			logger.error("Exception in IndexerQueueTaskBuilderIbm " + e.toString());
+			e.printStackTrace();
+		}
+
+	}
+}
\ No newline at end of file

provider/indexer-ibm/src/main/resources/application.properties

@@ -64,3 +64,7 @@ ELASTIC_DATASTORE_ID=indexer-service
 ELASTIC_HOST=elasticsearch.com
 ELASTIC_PORT=443
 ELASTIC_USER_PASSWORD=REPLACE_ME:REPLACE_ME
+
+#requeue logic
+INDEXER_QUEUE_HOST=""
+ibm.env.prefix=oc-local
\ No newline at end of file