Merge remote-tracking branch 'origin/master' into integration-master

# Conflicts:
#	indexer-core/src/main/java/org/opengroup/osdu/indexer/api/CleanupIndiciesApi.java
#	indexer-core/src/test/java/org/opengroup/osdu/indexer/api/CleanupIndiciesApiTest.java
#	provider/indexer-gcp/src/main/appengine/app.yaml
#	provider/indexer-gcp/src/main/resources/application.properties

.gitlab-ci.yml

@@ -14,6 +14,14 @@ variables:
   GCP_DOMAIN: cloud.slb-ds.com
   GCP_STORAGE_URL: https://osdu-indexer-dot-opendes.appspot.com/api/storage/v2/
 
+  OSDU_GCP_BUILD_SUBDIR: provider/indexer-gcp
+  OSDU_GCP_INT_TEST_SUBDIR: testing/indexer-test-gcp
+  OSDU_GCP_APPLICATION_NAME: os-indexer
+  OSDU_GCP_PROJECT: nice-etching-277309
+  OSDU_GCP_TENANT_NAME: osdu
+  OSDU_GCP_STORAGE_SCHEMA_HOST: https://os-storage-dot-nice-etching-277309.uc.r.appspot.com/api/storage/v2/schemas
+  OSDU_SECURITY_HTTPS_CERTIFICATE_TRUST: 'true'
+
   IBM_BUILD_SUBDIR: provider/indexer-ibm
   IBM_INT_TEST_SUBDIR: testing/indexer-test-ibm
 
@@ -46,5 +54,10 @@ include:
   - project: "osdu/platform/ci-cd-pipelines"
     file: "publishing/pages.yml"
 
+  - project: 'osdu/platform/ci-cd-pipelines'
+    ref: "master"
+    file: 'cloud-providers/osdu-gcp.yml'
+
+
 aws-test-java:
   tags: ['aws-internal-test']

NOTICE

@@ -11,7 +11,6 @@ The following software have components provided under the terms of this license:
 - Cobertura code coverage (from http://cobertura.sourceforge.net)
 - Plexus :: Default Container (from )
 - Plexus Common Utilities (from http://plexus.codehaus.org/plexus-utils)
-- StAX (from http://stax.codehaus.org/)
 - oro (from )
 
 ========================================================================
@@ -244,6 +243,10 @@ The following software have components provided under the terms of this license:
 - Guava: Google Core Libraries for Java (from https://github.com/google/guava.git)
 - HPPC Collections (from http://labs.carrotsearch.com)
 - Hibernate Validator Engine (from )
+- IBM COS Java SDK for Amazon S3 (from https://github.com/ibm/ibm-cos-sdk-java)
+- IBM COS Java SDK for COS KMS (from https://github.com/ibm/ibm-cos-sdk-java)
+- IBM COS SDK For Java (from https://github.com/ibm/ibm-cos-sdk-java)
+- IBM COS SDK for Java - Core (from https://github.com/ibm/ibm-cos-sdk-java)
 - Identity and Access Management (IAM) API v1-rev247-1.23.0 (from )
 - J2ObjC Annotations (from https://github.com/google/j2objc/)
 - J2ObjC Annotations (from https://github.com/google/j2objc/)
@@ -336,9 +339,9 @@ The following software have components provided under the terms of this license:
 - Microsoft Application Insights Log4j 2 Appender (from https://github.com/Microsoft/ApplicationInsights-Java)
 - Microsoft Azure Netty HTTP Client Library (from https://github.com/Azure/azure-sdk-for-java)
 - Mockito (from http://mockito.org)
-- Mockito (from http://mockito.org)
 - Mockito (from http://www.mockito.org)
 - Mockito (from http://mockito.org)
+- Mockito (from http://mockito.org)
 - Mojo's Maven plugin for Cobertura (from http://mojo.codehaus.org/cobertura-maven-plugin/)
 - Netty Reactive Streams Implementation (from )
 - Netty/Buffer (from http://netty.io/)
@@ -368,7 +371,6 @@ The following software have components provided under the terms of this license:
 - OkHttp URLConnection (from )
 - OkHttp URLConnection (from )
 - Okio (from )
-- Okio (from )
 - OpenCensus (from https://github.com/census-instrumentation/opencensus-java)
 - OpenCensus (from https://github.com/census-instrumentation/opencensus-java)
 - OpenCensus (from https://github.com/census-instrumentation/opencensus-java)
@@ -385,7 +387,6 @@ The following software have components provided under the terms of this license:
 - Reactive Streams Netty driver (from https://github.com/reactor/reactor-netty)
 - Retrofit (from )
 - Servlet Specification 2.5 API (from )
-- Simple XML (from http://simple.sourceforge.net)
 - SnakeYAML (from http://www.snakeyaml.org)
 - Spatial4J (from http://www.locationtech.org/projects/locationtech.spatial4j)
 - Spring AOP (from https://github.com/spring-projects/spring-framework)
@@ -425,8 +426,6 @@ The following software have components provided under the terms of this license:
 - Spring Transaction (from https://github.com/spring-projects/spring-framework)
 - Spring Web (from https://github.com/spring-projects/spring-framework)
 - Spring Web MVC (from https://github.com/spring-projects/spring-framework)
-- StAX (from http://stax.codehaus.org/)
-- StAX API (from http://stax.codehaus.org/)
 - T-Digest (from https://github.com/tdunning/t-digest)
 - Woodstox (from https://github.com/FasterXML/woodstox)
 - Xerces2-j (from https://xerces.apache.org/xerces2-j/)
@@ -457,6 +456,7 @@ The following software have components provided under the terms of this license:
 - io.grpc:grpc-protobuf-lite (from https://github.com/grpc/grpc-java)
 - io.grpc:grpc-stub (from https://github.com/grpc/grpc-java)
 - ion-java (from https://github.com/amznlabs/ion-java/)
+- ion-java (from https://github.com/amznlabs/ion-java/)
 - jackson-databind (from http://github.com/FasterXML/jackson)
 - java-cloudant (from https://cloudant.com)
 - java-cloudant (from https://cloudant.com)
@@ -479,7 +479,6 @@ The following software have components provided under the terms of this license:
 - lettuce (from http://github.com/mp911de/lettuce/wiki)
 - micrometer-core (from https://github.com/micrometer-metrics/micrometer)
 - micrometer-registry-azure-monitor (from https://github.com/micrometer-metrics/micrometer)
-- minio (from https://github.com/minio/minio-java)
 - org.xmlunit:xmlunit-core (from http://www.xmlunit.org/)
 - oro (from )
 - parent-join (from https://github.com/elastic/elasticsearch)
@@ -542,7 +541,6 @@ The following software have components provided under the terms of this license:
 - Lucene Common Analyzers (from )
 - Plexus :: Default Container (from )
 - Plexus Common Utilities (from http://plexus.codehaus.org/plexus-utils)
-- StAX (from http://stax.codehaus.org/)
 - Stax2 API (from http://github.com/FasterXML/stax2-api)
 - jersey-ext-bean-validation (from )
 - jersey-spring4 (from )
@@ -616,7 +614,6 @@ CC-BY-3.0
 ========================================================================
 The following software have components provided under the terms of this license:
 
-- "Java Concurrency in Practice" book annotations (from http://jcip.net/)
 - FindBugs-jsr305 (from http://findbugs.sourceforge.net/)
 
 ========================================================================
@@ -813,7 +810,6 @@ The following software have components provided under the terms of this license:
 - Microsoft Application Insights Java SDK Spring Boot starter (from https://github.com/Microsoft/ApplicationInsights-Java)
 - Microsoft Application Insights Java SDK Web Module (from https://github.com/Microsoft/ApplicationInsights-Java)
 - Microsoft Application Insights Log4j 2 Appender (from https://github.com/Microsoft/ApplicationInsights-Java)
-- SpotBugs Annotations (from https://spotbugs.github.io/)
 - java-getopt (from )
 
 ========================================================================
@@ -825,7 +821,6 @@ The following software have components provided under the terms of this license:
 - Java Native Access Platform (from https://github.com/java-native-access/jna)
 - Javassist (from http://www.javassist.org/)
 - SnakeYAML (from http://www.snakeyaml.org)
-- SpotBugs Annotations (from https://spotbugs.github.io/)
 
 ========================================================================
 LGPL-3.0-only
@@ -875,10 +870,10 @@ The following software have components provided under the terms of this license:
 - Microsoft Azure client library for Identity (from https://github.com/Azure/azure-sdk-for-java)
 - Microsoft Azure client library for KeyVault Secrets (from https://github.com/Azure/azure-sdk-for-java)
 - Microsoft Azure common module for Storage (from https://github.com/Azure/azure-sdk-for-java)
+- Mockito (from http://www.mockito.org)
 - Mockito (from http://mockito.org)
 - Mockito (from http://mockito.org)
 - Mockito (from http://mockito.org)
-- Mockito (from http://www.mockito.org)
 - Netty/Codec/HTTP (from )
 - Netty/Common (from )
 - Plexus :: Default Container (from )

devops/Jenkinsfile

+pipeline {
+ agent {
+   kubernetes {
+   cloud 'openshift'
+     label 'maven-openjdk18'
+	 yaml """
+spec:
+  containers:
+  - name: jnlp
+    image: quay.io/openshift/origin-jenkins-agent-maven:v4.0.0
+    volumeMounts:
+    - mountPath: "/tmp"
+      name: "workspace-volume"
+      readOnly: false
+    workingDir: "/tmp"
+    securityContext:
+      privileged: false	
+    tty: false
+    resources:
+      limits: 
+        cpu: 200m
+        memory: 2Gi
+      requests: 
+        cpu: 200m
+        memory: 2Gi    
+  restartPolicy: "Never"
+"""		 
+   }
+ }
+
+ environment { 
+  
+ //Cluster environment variable(CLS_ENV). Like QA, DEV, PERF, PROD etc.
+ CLS_ENV = "dev"
+ 
+ //Service variable(CORE_SERVICE). Like indexer, search, delivery, storage, legal etc.
+ CORE_SERVICE = "indexer"
+
+ //GitHub repo URL credential ID for Environment variable files which saved as Secure text in Jenkins Credential.
+ GIT_ENV_VAR_PATH_URL =  credentials('GitRepo-URL-For-Environment-variables') 
+ 
+ //Personal token variable ID which saved as Secure text in Jenkins Credential. Like: GitHub-PRIVATE-TOKEN.
+ PRIVATE_TOKEN  = credentials('GitHub-PRIVATE-TOKEN')
+ 
+ def runShell =  sh (returnStdout: true, script: "curl --header 'PRIVATE-TOKEN: $PRIVATE_TOKEN' ''$GIT_ENV_VAR_PATH_URL'%2F'$CORE_SERVICE'_'$CLS_ENV'_env.json/raw?ref=master' -s -o env.json")
+
+ }
+
+   stages {
+      stage('Integration_test') {
+          environment {
+            def readContent = readJSON file: 'env.json'
+
+			AUTH_USER_ACCESS = "${readContent['AUTH_USER_ACCESS']}"
+			AUTH_USER_ACCESS_PASSWORD = "${readContent['AUTH_USER_ACCESS_PASSWORD']}"
+			DEFAULT_DATA_PARTITION_ID_TENANT1 = "${readContent['DEFAULT_DATA_PARTITION_ID_TENANT1']}"
+			DEFAULT_DATA_PARTITION_ID_TENANT2 = "${readContent['DEFAULT_DATA_PARTITION_ID_TENANT2']}"
+			ELASTIC_HOST = "${readContent['ELASTIC_HOST']}"
+			ELASTIC_PASSWORD = "${readContent['ELASTIC_PASSWORD']}"
+			ELASTIC_PORT = "${readContent['ELASTIC_PORT']}"
+			ELASTIC_USER_NAME = "${readContent['ELASTIC_USER_NAME']}"
+			ENTITLEMENTS_DOMAIN = "${readContent['ENTITLEMENTS_DOMAIN']}"
+			KEYCLOAK_CLIENT_ID = "${readContent['KEYCLOAK_CLIENT_ID']}"
+			KEYCLOAK_CLIENT_SECRET = "${readContent['KEYCLOAK_CLIENT_SECRET']}"
+			KEYCLOAK_REALM = "${readContent['KEYCLOAK_REALM']}"
+			KEYCLOAK_URL = "${readContent['KEYCLOAK_URL']}"
+			LEGAL_TAG = "${readContent['LEGAL_TAG']}"
+			OTHER_RELEVANT_DATA_COUNTRIES = "${readContent['OTHER_RELEVANT_DATA_COUNTRIES']}"
+			STORAGE_HOST = "${readContent['STORAGE_HOST']}"
+			SEARCH_HOST = "${readContent['SEARCH_HOST']}"
+			LEGAL_HOST = "${readContent['LEGAL_HOST']}"
+			INDEXER_HOST = "${readContent['INDEXER_HOST']}"
+			ENTITLEMENT_URL = "${readContent['ENTITLEMENT_URL']}"
+
+          }  
+            steps {
+               script {
+                  sh 'mvn -f testing/indexer-test-ibm/pom.xml test'
+               }
+            }
+       }
+    }
+
+
+}
\ No newline at end of file

indexer-core/src/main/java/org/opengroup/osdu/indexer/api/CleanupIndiciesApi.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.api;
 
 import com.google.common.reflect.TypeToken;
 import com.google.gson.Gson;
 import com.google.gson.JsonParseException;
-import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.List;
-import java.util.Map;
 import javax.validation.Valid;
 import javax.validation.constraints.NotNull;
 import lombok.extern.java.Log;
-import org.elasticsearch.ElasticsearchStatusException;
-import org.elasticsearch.client.RestHighLevelClient;
 import org.opengroup.osdu.core.common.model.http.AppException;
 import org.opengroup.osdu.core.common.model.http.DpsHeaders;
-import org.opengroup.osdu.core.common.model.indexer.OperationType;
 import org.opengroup.osdu.core.common.model.indexer.RecordInfo;
 import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
 import org.opengroup.osdu.core.common.model.search.SearchServiceRole;
-import org.opengroup.osdu.core.common.search.ElasticIndexNameResolver;
-import org.opengroup.osdu.core.common.search.IndicesService;
 import org.opengroup.osdu.indexer.SwaggerDoc;
-import org.opengroup.osdu.indexer.util.ElasticClientHandler;
+import org.opengroup.osdu.indexer.service.IndexerService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -37,37 +44,31 @@ import org.springframework.web.context.annotation.RequestScope;
 public class CleanupIndiciesApi {
 
   @Autowired
-  private ElasticClientHandler elasticClientHandler;
-
-  @Autowired
-  private ElasticIndexNameResolver elasticIndexNameResolver;
-
-  @Autowired
-  private IndicesService indicesService;
+  private IndexerService indexerService;
 
   @PostMapping(path = "/index-cleanup", consumes = "application/json")
-  @PreAuthorize("@authorizationFilter.hasRole('" + SearchServiceRole.ADMIN + "')")
+  @PreAuthorize("@authorizationFilter.hasPermission('" + SearchServiceRole.ADMIN + "')")
   public ResponseEntity cleanupIndices(@NotNull(message = SwaggerDoc.REQUEST_VALIDATION_NOT_NULL_BODY)
-                                         @Valid @RequestBody RecordChangedMessages recordChangedMessages) throws Exception {
+                                         @Valid @RequestBody RecordChangedMessages message) {
+    if (message == null) {
+      throw new AppException(HttpStatus.BAD_REQUEST.value(), "Request body is null",
+          SwaggerDoc.REQUEST_VALIDATION_NOT_NULL_BODY);
+    }
 
-    if (recordChangedMessages.missingAccountId()) {
-      throw new AppException(org.apache.http.HttpStatus.SC_BAD_REQUEST, "Invalid tenant",
+    if (message.missingAccountId()) {
+      throw new AppException(HttpStatus.BAD_REQUEST.value(), "Invalid tenant",
           String.format("Required header: '%s' not found", DpsHeaders.DATA_PARTITION_ID));
     }
     try {
-      if (recordChangedMessages == null) {
-        log.info("record change messages is null");
-      }
-
       Type listType = new TypeToken<List<RecordInfo>>() {
       }.getType();
-      List<RecordInfo> recordInfos = new Gson().fromJson(recordChangedMessages.getData(), listType);
+      List<RecordInfo> recordInfos = new Gson().fromJson(message.getData(), listType);
 
-      if (recordInfos.size() == 0) {
+      if (recordInfos.isEmpty()) {
         log.info("none of record-change message can be deserialized");
         return new ResponseEntity(HttpStatus.OK);
       }
-      processSchemaMessages(recordChangedMessages, recordInfos);
+      indexerService.processSchemaMessages(recordInfos);
       return new ResponseEntity(HttpStatus.OK);
     } catch (AppException e) {
       throw e;
@@ -77,34 +78,4 @@ public class CleanupIndiciesApi {
       throw new AppException(HttpStatus.BAD_REQUEST.value(), "Unknown error", "An unknown error has occurred.", e);
     }
   }
-
-  private void processSchemaMessages(RecordChangedMessages message, List<RecordInfo> recordInfos) throws Exception {
-    Map<String, OperationType> schemaMsgs = RecordInfo.getSchemaMsgs(recordInfos);
-    if (schemaMsgs != null && !schemaMsgs.isEmpty()) {
-      try (RestHighLevelClient restClient = elasticClientHandler.createRestClient()) {
-        schemaMsgs.entrySet().forEach(msg -> {
-          try {
-            processSchemaEvents(restClient, msg);
-          } catch (IOException | ElasticsearchStatusException e) {
-            throw new AppException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "unable to process schema delete", e.getMessage());
-          }
-        });
-      }
-    }
-  }
-
-  private void processSchemaEvents(RestHighLevelClient restClient,
-                                   Map.Entry<String, OperationType> msg) throws IOException, ElasticsearchStatusException {
-    String kind = msg.getKey();
-    String index = elasticIndexNameResolver.getIndexNameFromKind(kind);
-
-    boolean indexExist = indicesService.isIndexExist(restClient, index);
-    if (msg.getValue() == OperationType.purge_schema) {
-      if (indexExist) {
-        indicesService.deleteIndex(restClient, index);
-      } else {
-        log.warning(String.format("Kind: %s not found", kind));
-      }
-    }
-  }
 }

indexer-core/src/main/java/org/opengroup/osdu/indexer/service/IndexerService.java

@@ -14,6 +14,7 @@
 
 package org.opengroup.osdu.indexer.service;
 
+import java.io.IOException;
 import java.util.List;
 
 import org.opengroup.osdu.core.common.model.indexer.JobStatus;
@@ -24,4 +25,6 @@ public interface IndexerService {
 
     JobStatus processRecordChangedMessages(RecordChangedMessages recordChangedMessages, List<RecordInfo> recordInfos) throws Exception;
 
+    void processSchemaMessages(List<RecordInfo> recordInfos) throws IOException;
+
 }
\ No newline at end of file

indexer-core/src/main/java/org/opengroup/osdu/indexer/service/IndexerServiceImpl.java

@@ -153,6 +153,33 @@ public class IndexerServiceImpl implements IndexerService {
         return jobStatus;
     }
 
+    @Override
+    public void processSchemaMessages(List<RecordInfo> recordInfos) throws IOException {
+        Map<String, OperationType> schemaMsgs = RecordInfo.getSchemaMsgs(recordInfos);
+        if (schemaMsgs != null && !schemaMsgs.isEmpty()) {
+            try (RestHighLevelClient restClient = elasticClientHandler.createRestClient()) {
+                schemaMsgs.entrySet().forEach(msg -> {
+                    try {
+                        processSchemaEvents(restClient, msg);
+                    } catch (IOException | ElasticsearchStatusException e) {
+                        throw new AppException(org.springframework.http.HttpStatus.INTERNAL_SERVER_ERROR.value(), "unable to process schema delete", e.getMessage());
+                    }
+                });
+            }
+        }
+    }
+
+    private void processSchemaEvents(RestHighLevelClient restClient,
+                                     Map.Entry<String, OperationType> msg) throws IOException, ElasticsearchStatusException {
+        String kind = msg.getKey();
+        String index = elasticIndexNameResolver.getIndexNameFromKind(kind);
+
+        boolean indexExist = indicesService.isIndexExist(restClient, index);
+        if (indexExist && msg.getValue() == OperationType.purge_schema) {
+            indicesService.deleteIndex(restClient, index);
+        }
+    }
+
     private List<String> processUpsertRecords(Map<String, Map<String, OperationType>> upsertRecordMap) throws Exception {
         // get schema for kind
         Map<String, IndexSchema> schemas = this.getSchema(upsertRecordMap);

indexer-core/src/test/java/org/opengroup/osdu/indexer/api/CleanupIndiciesApiTest.java

@@ -10,15 +10,11 @@ import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.opengroup.osdu.core.common.http.HeadersUtil;
-import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
 import org.opengroup.osdu.core.common.model.http.AppException;
 import org.opengroup.osdu.core.common.model.http.DpsHeaders;
 import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
 import org.opengroup.osdu.core.common.search.Config;
-import org.opengroup.osdu.core.common.search.ElasticIndexNameResolver;
-import org.opengroup.osdu.core.common.search.IndicesService;
 import org.opengroup.osdu.indexer.service.IndexerService;
-import org.opengroup.osdu.indexer.util.ElasticClientHandler;
 import org.opengroup.osdu.indexer.util.IndexerQueueTaskBuilder;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
@@ -34,23 +30,11 @@ public class CleanupIndiciesApiTest {
   private final String messageWithEmptyData = "{\"data\":\"[]\",\"attributes\":{\"account-id\":\"opendes\",\"correlation-id\":\"b5a281bd-f59d-4db2-9939-b2d85036fc7e\"},\"messageId\":\"75328163778221\",\"publishTime\":\"2018-05-08T21:48:56.131Z\"}";
   private final String messageWithIncorrectJsonFormat = "{\"data\":\"[{}}]\",\"attributes\":{\"account-id\":\"opendes\",\"correlation-id\":\"b5a281bd-f59d-4db2-9939-b2d85036fc7e\"},\"messageId\":\"75328163778221\",\"publishTime\":\"2018-05-08T21:48:56.131Z\"}";
 
-  private final String ACCOUNT_ID = "any-account";
-  private final String DATA_PARTITION_ID = "opendes";
-
   @InjectMocks
   private CleanupIndiciesApi sut;
 
   @Mock
-  private JaxRsDpsLog log;
-
-  @Mock
-  private ElasticClientHandler elasticClientHandler;
-
-  @Mock
-  private ElasticIndexNameResolver elasticIndexNameResolver;
-
-  @Mock
-  private IndicesService indicesService;
+  private IndexerService indexerService;
 
   @Before
   public void setup() {
@@ -58,12 +42,12 @@ public class CleanupIndiciesApiTest {
   }
 
   @Test
-  public void should_return200_given_validMessage_indexCleanupTest() throws Exception {
+  public void should_return200_given_validMessage_indexCleanupTest() {
     should_return200_indexerWorkerTest(messageValid);
   }
 
   @Test
-  public void should_return200_given_emptyData_indexCleanupTest() throws Exception {
+  public void should_return200_given_emptyData_indexCleanupTest() {
     should_return200_indexerWorkerTest(messageWithEmptyData);
   }
 
@@ -77,7 +61,7 @@ public class CleanupIndiciesApiTest {
     should_return400_indexerWorkerTest(messageWithIncorrectJsonFormat, "Unable to parse request payload.");
   }
 
-  private void should_return200_indexerWorkerTest(String message) throws Exception {
+  private void should_return200_indexerWorkerTest(String message) {
     ResponseEntity response = this.sut.cleanupIndices(createRecordChangedMessage(message));
     Assert.assertEquals(HttpStatus.OK.value(), response.getStatusCodeValue());
   }

indexer-core/src/test/java/org/opengroup/osdu/indexer/service/IndexerServiceImplTest.java

+package org.opengroup.osdu.indexer.service;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.MockitoAnnotations.initMocks;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.opengroup.osdu.core.common.model.indexer.RecordInfo;
+import org.opengroup.osdu.core.common.search.ElasticIndexNameResolver;
+import org.opengroup.osdu.core.common.search.IndicesService;
+import org.opengroup.osdu.indexer.util.ElasticClientHandler;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+public class IndexerServiceImplTest {
+
+  @InjectMocks
+  private IndexerServiceImpl indexerService;
+
+  @Mock
+  private ElasticClientHandler elasticClientHandler;
+
+  @Mock
+  private ElasticIndexNameResolver elasticIndexNameResolver;
+
+  @Mock
+  private IndicesService indicesService;
+
+  private List<RecordInfo> recordInfos = new ArrayList<>();
+
+  @Before
+  public void setup() {
+    RecordInfo recordInfo = new RecordInfo();
+    recordInfo.setId("opendes:ds:mytest3-d9033ae1-fb15-496c-9ba0-880fd1d2b2qf");
+    recordInfo.setKind("opendes:ds:mytest2:1.0.0");
+    recordInfo.setOp("purge_schema");
+    recordInfos.add(recordInfo);
+
+    initMocks(this);
+  }
+
+  @Test
+  public void processSchemaMessagesTest() throws Exception {
+    indexerService.processSchemaMessages(recordInfos);
+
+    verify(elasticClientHandler, times(1)).createRestClient();
+    verify(elasticIndexNameResolver, times(1)).getIndexNameFromKind(any());
+    verify(indicesService, times(1)).isIndexExist(any(), any());
+  }
+}

provider/indexer-gcp/src/main/appengine/app.yaml

@@ -39,4 +39,4 @@ env_variables:
   LEGAL_HOSTNAME: "LEGAL_HOSTNAME_VAR"
   REGION: "REGION_VAR"
   SPRING_PROFILES_ACTIVE: 'ENVIRONMENT'
-  SECURITY_HTTPS_CERTIFICATE_TRUST: 'SECURITY_HTTPS_CERTIFICATE_TRUST_VAR'
\ No newline at end of file
+  SECURITY_HTTPS_CERTIFICATE_TRUST: 'SECURITY_HTTPS_CERTIFICATE_TRUST_VAR'

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

@@ -33,4 +33,4 @@ KMS_KEY=searchService
 ELASTIC_DATASTORE_KIND=SearchSettings
 ELASTIC_DATASTORE_ID=indexer-service
 
-security.https.certificate.trust=${SECURITY_HTTPS_CERTIFICATE_TRUST}
+security.https.certificate.trust=false

provider/indexer-ibm/pom.xml

@@ -33,7 +33,7 @@
     <packaging>jar</packaging>
 	
 	<properties>
-		<os-core-lib-ibm.version>0.0.18</os-core-lib-ibm.version>
+		<os-core-lib-ibm.version>0.3.8-SNAPSHOT</os-core-lib-ibm.version>
 	</properties>
 	
 	<profiles>

provider/indexer-ibm/src/main/java/org/opengroup/osdu/indexer/ibm/security/SecurityConfig.java

@@ -29,6 +29,8 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter {
                 .csrf().disable()
                 .authorizeRequests()
                 .antMatchers("/", "/index.html",
+                		"/liveness_check",
+                		"/readiness_check",
                         "/index-worker", "/_dps/task-handlers", "/_dps/task-handlers/**",
                         "/reindex",
                         "/v2/api-docs",

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/java/org/opengroup/osdu/indexer/ibm/util/RequestInfoImpl.java

@@ -53,10 +53,14 @@ public class RequestInfoImpl implements IRequestInfo {
 
     @Inject
     private TenantInfo tenantInfo;
-
+    
     @Value("${DEPLOYMENT_ENVIRONMENT}")
     private String DEPLOYMENT_ENVIRONMENT;
 
+    private static final String INDEXER_API_KEY_HEADER="x-api-key";
+    
+    @Value("${INDEXER_API_KEY}")
+    private String tokenFromProperty;
 
     @Override
     public DpsHeaders getHeaders() {
@@ -65,9 +69,19 @@ public class RequestInfoImpl implements IRequestInfo {
             // throw to prevent null reference exception below
             throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Invalid Headers", "Headers Map DpsHeaders is null");
         }
-        DpsHeaders headers = this.getCoreServiceHeaders(headersMap.getHeaders());
-        return headers;
-    }
+		DpsHeaders headers = this.getCoreServiceHeaders(headersMap.getHeaders());
+
+		if (headers.getHeaders().containsKey(INDEXER_API_KEY_HEADER)) {
+			String apiToken = headers.getHeaders().get(INDEXER_API_KEY_HEADER);
+			if (!apiToken.equals(tokenFromProperty)) {
+				logger.error("Indexer API Token in header is mismatched");
+				throw new AppException(HttpStatus.SC_UNAUTHORIZED, "Indexer API Token in header mismatched.", "Indexer API Token in header mismatched.");
+			}
+		} else {
+			throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Missing Header", "The headers "+ INDEXER_API_KEY_HEADER + "  is missing!");
+		}
+		return headers;
+	}
 
     @Override
     public String getPartitionId() {
@@ -95,7 +109,8 @@ public class RequestInfoImpl implements IRequestInfo {
 
     @Override
     public boolean isTaskQueueRequest() {
-        //if (!this.dpsHeaders.getHeaders().containsKey(INDEXER_QUEUE_KEY)) return false;
+        //if (!this.dpsHeaders.getHeaders().containsKey(INDEXER_API_KEY_HEADER)) return false;
+    	
 
 //        String queueId = this.headersInfo.getHeadersMap().get(AppEngineHeaders.TASK_QUEUE_NAME);
 //        return queueId.endsWith(Constants.INDEXER_QUEUE_IDENTIFIER);

testing/indexer-test-core/src/main/resources/features/indexrecord/IndexRecord.feature

@@ -15,8 +15,8 @@ Feature: Indexing of the documents
 
     Examples:
       | kind                                      | recordFile        | number | index                                     | type   | acl                                    | mapping                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
-      | "tenant1:testindex<timestamp>:well:1.0.0" | "index_records_1" | 5      | "tenant1-testindex<timestamp>-well-1.0.0" | "well" | "data.default.viewers@opendes" | "{"mappings":{"well":{"dynamic":"false","properties":{"acl":{"properties":{"owners":{"type":"keyword"},"viewers":{"type":"keyword"}}},"ancestry":{"properties":{"parents":{"type":"keyword"}}},"data":{"properties":{"Basin":{"type":"text"},"Country":{"type":"text"},"County":{"type":"text"},"EmptyAttribute":{"type":"text"},"Established":{"type":"date"},"Field":{"type":"text"},"Location":{"type":"geo_point"},"OriginalOperator":{"type":"text"},"Rank":{"type":"integer"},"Score":{"type":"integer"},"State":{"type":"text"},"WellName":{"type":"text"},"WellStatus":{"type":"text"},"WellType":{"type":"text"},"DblArray":{"type":"double"}}},"id":{"type":"keyword"},"index":{"properties":{"lastUpdateTime":{"type":"date"},"statusCode":{"type":"integer"},"trace":{"type":"text"}}},"kind":{"type":"keyword"},"legal":{"properties":{"legaltags":{"type":"keyword"},"otherRelevantDataCountries":{"type":"keyword"},"status":{"type":"keyword"}}},"namespace":{"type":"keyword"},"type":{"type":"keyword"},"version":{"type":"long"},"x-acl":{"type":"keyword"}}}}}" |
-      | "tenant1:testindex<timestamp>:well:3.0.0" | "index_records_1" | 5      | "tenant1-testindex<timestamp>-well-3.0.0" | "well" | "data.default.viewers@opendes" | "{"mappings":{"well":{"dynamic":"false","properties":{"acl":{"properties":{"owners":{"type":"keyword"},"viewers":{"type":"keyword"}}},"ancestry":{"properties":{"parents":{"type":"keyword"}}},"data":{"properties":{"Basin":{"type":"text"},"Country":{"type":"text"},"County":{"type":"text"},"EmptyAttribute":{"type":"text"},"Established":{"type":"date"},"Field":{"type":"text"},"Location":{"type":"geo_point"},"OriginalOperator":{"type":"text"},"Rank":{"type":"integer"},"Score":{"type":"integer"},"State":{"type":"text"},"WellName":{"type":"text"},"WellStatus":{"type":"text"},"WellType":{"type":"text"},"DblArray":{"type":"double"}}},"id":{"type":"keyword"},"index":{"properties":{"lastUpdateTime":{"type":"date"},"statusCode":{"type":"integer"},"trace":{"type":"text"}}},"kind":{"type":"keyword"},"legal":{"properties":{"legaltags":{"type":"keyword"},"otherRelevantDataCountries":{"type":"keyword"},"status":{"type":"keyword"}}},"namespace":{"type":"keyword"},"type":{"type":"keyword"},"version":{"type":"long"},"x-acl":{"type":"keyword"}}}}}" |
+      | "tenant1:testindex<timestamp>:well:1.0.0" | "index_records_1" | 5      | "tenant1-testindex<timestamp>-well-1.0.0" | "well" | "data.default.viewers@tenant1" | "{"mappings":{"well":{"dynamic":"false","properties":{"acl":{"properties":{"owners":{"type":"keyword"},"viewers":{"type":"keyword"}}},"ancestry":{"properties":{"parents":{"type":"keyword"}}},"data":{"properties":{"Basin":{"type":"text"},"Country":{"type":"text"},"County":{"type":"text"},"EmptyAttribute":{"type":"text"},"Established":{"type":"date"},"Field":{"type":"text"},"Location":{"type":"geo_point"},"OriginalOperator":{"type":"text"},"Rank":{"type":"integer"},"Score":{"type":"integer"},"State":{"type":"text"},"WellName":{"type":"text"},"WellStatus":{"type":"text"},"WellType":{"type":"text"},"DblArray":{"type":"double"}}},"id":{"type":"keyword"},"index":{"properties":{"lastUpdateTime":{"type":"date"},"statusCode":{"type":"integer"},"trace":{"type":"text"}}},"kind":{"type":"keyword"},"legal":{"properties":{"legaltags":{"type":"keyword"},"otherRelevantDataCountries":{"type":"keyword"},"status":{"type":"keyword"}}},"namespace":{"type":"keyword"},"type":{"type":"keyword"},"version":{"type":"long"},"x-acl":{"type":"keyword"}}}}}" |
+      | "tenant1:testindex<timestamp>:well:3.0.0" | "index_records_1" | 5      | "tenant1-testindex<timestamp>-well-3.0.0" | "well" | "data.default.viewers@tenant1" | "{"mappings":{"well":{"dynamic":"false","properties":{"acl":{"properties":{"owners":{"type":"keyword"},"viewers":{"type":"keyword"}}},"ancestry":{"properties":{"parents":{"type":"keyword"}}},"data":{"properties":{"Basin":{"type":"text"},"Country":{"type":"text"},"County":{"type":"text"},"EmptyAttribute":{"type":"text"},"Established":{"type":"date"},"Field":{"type":"text"},"Location":{"type":"geo_point"},"OriginalOperator":{"type":"text"},"Rank":{"type":"integer"},"Score":{"type":"integer"},"State":{"type":"text"},"WellName":{"type":"text"},"WellStatus":{"type":"text"},"WellType":{"type":"text"},"DblArray":{"type":"double"}}},"id":{"type":"keyword"},"index":{"properties":{"lastUpdateTime":{"type":"date"},"statusCode":{"type":"integer"},"trace":{"type":"text"}}},"kind":{"type":"keyword"},"legal":{"properties":{"legaltags":{"type":"keyword"},"otherRelevantDataCountries":{"type":"keyword"},"status":{"type":"keyword"}}},"namespace":{"type":"keyword"},"type":{"type":"keyword"},"version":{"type":"long"},"x-acl":{"type":"keyword"}}}}}" |
 
   Scenario Outline: Ingest the record and Index in the Elastic Search with bad attribute
     When I ingest records with the <recordFile> with <acl> for a given <kind>
@@ -24,5 +24,5 @@ Feature: Indexing of the documents
 
     Examples:
       | kind                                      | recordFile        | number | index                                     | skippedAttribute | acl                            |
-      | "tenant1:testindex<timestamp>:well:2.0.0" | "index_records_2" | 4      | "tenant1-testindex<timestamp>-well-2.0.0" | "data.Location"  | "data.default.viewers@opendes" |
-      | "tenant1:testindex<timestamp>:well:3.0.0" | "index_records_3" | 7      | "tenant1-testindex<timestamp>-well-3.0.0" | "data.GeoShape"  | "data.default.viewers@opendes" |
+      | "tenant1:testindex<timestamp>:well:2.0.0" | "index_records_2" | 4      | "tenant1-testindex<timestamp>-well-2.0.0" | "data.Location"  | "data.default.viewers@tenant1" |
+      | "tenant1:testindex<timestamp>:well:3.0.0" | "index_records_3" | 7      | "tenant1-testindex<timestamp>-well-3.0.0" | "data.GeoShape"  | "data.default.viewers@tenant1" |

testing/indexer-test-ibm/pom.xml

@@ -14,7 +14,7 @@
         <maven.compiler.target>1.8</maven.compiler.target>
         <maven.compiler.source>1.8</maven.compiler.source>
         <cucumber.version>1.2.5</cucumber.version>
-        <os-core-lib-ibm.version>0.0.18</os-core-lib-ibm.version>
+        <os-core-lib-ibm.version>0.3.8-SNAPSHOT</os-core-lib-ibm.version>
     </properties>
 
     <dependencies>

testing/indexer-test-ibm/src/main/resources/apikey.feature

+Feature: check the api key in header
+  This feature check the api key received from headers is matching with configured in environment
+
+  Scenario Outline: compare the api key with configured environment key
+    When I pass api key
+    Then compare with key configured in properties file
+
+    Examples:
+      | apiKey    | 
+      | "abcd" 	  | 

testing/indexer-test-ibm/src/test/java/org/opengroup/osdu/step_definitions/index/record/RunTest.java

@@ -6,7 +6,7 @@ import org.junit.runner.RunWith;
 
 @RunWith(Cucumber.class)
 @CucumberOptions(
-        features = "classpath:features/indexrecord/IndexRecord.feature",
+        features = {"classpath:features/indexrecord/IndexRecord.feature","classpath:apikey.feature"},
         glue = {"classpath:org.opengroup.osdu.step_definitions/index/record"},
         plugin = {"pretty", "junit:target/cucumber-reports/TEST-indexrecord.xml"})
 public class RunTest {

testing/indexer-test-ibm/src/test/java/org/opengroup/osdu/step_definitions/index/record/Steps.java

 package org.opengroup.osdu.step_definitions.index.record;
 
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
 import org.opengroup.osdu.common.RecordSteps;
+import org.opengroup.osdu.core.common.Constants;
+import org.opengroup.osdu.core.common.http.HttpClient;
+import org.opengroup.osdu.core.common.http.HttpRequest;
+import org.opengroup.osdu.core.common.http.HttpResponse;
+import org.opengroup.osdu.core.common.model.http.AppError;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
+import org.opengroup.osdu.core.ibm.util.Config;
 import org.opengroup.osdu.util.IBMHTTPClient;
 
+import com.google.gson.Gson;
+
 import cucumber.api.DataTable;
 import cucumber.api.Scenario;
 import cucumber.api.java.Before;
@@ -48,5 +65,46 @@ public class Steps extends RecordSteps {
     public void iShouldGetTheNumberDocumentsForTheIndexInTheElasticSearchWithOutSkippedAttribute(int expectedCount, String index, String skippedAttributes) throws Throwable {
         super.iShouldGetTheNumberDocumentsForTheIndexInTheElasticSearchWithOutSkippedAttribute(expectedCount, index, skippedAttributes);
     }
-
+    
+    @When("^I pass api key$")
+    public void i_pass_the_api_key() {
+    }
+    
+    @Then("^compare with key configured in properties file$")
+    public void compare_with_key_configured_in_propertiesFile() throws Throwable {
+    	
+    	final String CORRELATION_ID = "1234";
+        final String OPENDES = "opendes";
+    	final Gson gson = new Gson();
+    	
+    	String INDEXER_API_KEY = Config.getEnvironmentVariable("INDEXER_API_KEY");
+    	String INDEXER_HOST_URL = Config.getEnvironmentVariable("INDEXER_HOST_URL");
+    	
+    	RecordChangedMessages recordChangeMessage = new RecordChangedMessages();
+    	
+    	String data = "[{\"id\":\"opendes:doc:1234\",\"kind\":\"opendes:test:test:1.0.0\",\"op\":\"create\"}]";
+    	
+    	Map<String, String> attributes = new HashMap<>();
+    	attributes.put("correlation-id", CORRELATION_ID);
+    	attributes.put("data-partition-id", OPENDES);
+    	recordChangeMessage.setAttributes(attributes);
+    	recordChangeMessage.setData(data);
+    	
+    	String url = StringUtils.join(INDEXER_HOST_URL, Constants.WORKER_RELATIVE_URL);
+		HttpClient httpClient = new HttpClient();
+		DpsHeaders dpsHeaders = new DpsHeaders();
+		dpsHeaders.put("x-api-key", INDEXER_API_KEY);
+		dpsHeaders.put("correlation-id", CORRELATION_ID);
+		dpsHeaders.put("data-partition-id", OPENDES);
+		
+		HttpRequest rq = HttpRequest.post(recordChangeMessage).url(url).headers(dpsHeaders.getHeaders()).build();
+		HttpResponse result = httpClient.send(rq);
+		if(result.hasException().equals(false) && result.getResponseCode() == 500) {
+			assertTrue(result.getResponseCode() == 500);
+		} else {
+			AppError error = gson.fromJson(result.getBody(), AppError.class);
+			assertFalse("Token is mismatched", error.getCode() == 401);
+		}
+		
+    }
 }
\ No newline at end of file