Merge branch 'master' into jdk-17-migration

# Conflicts:
#	indexer-core/src/test/java/org/opengroup/osdu/indexer/api/RecordIndexerApiTest.java

indexer-core/src/main/java/org/opengroup/osdu/indexer/api/RecordIndexerApi.java

@@ -16,7 +16,6 @@ package org.opengroup.osdu.indexer.api;
 
 import com.google.common.reflect.TypeToken;
 import com.google.gson.Gson;
-
 import com.google.gson.JsonParseException;
 import io.swagger.v3.oas.annotations.Operation;
 import lombok.extern.java.Log;
@@ -28,14 +27,17 @@ import org.opengroup.osdu.core.common.model.indexer.RecordReindexRequest;
 import org.opengroup.osdu.core.common.model.indexer.SchemaChangedMessages;
 import org.opengroup.osdu.core.common.model.indexer.SchemaInfo;
 import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
+import org.opengroup.osdu.core.common.provider.interfaces.IRequestInfo;
 import org.opengroup.osdu.indexer.SwaggerDoc;
 import org.opengroup.osdu.indexer.service.IndexerService;
 import org.opengroup.osdu.indexer.service.ReindexService;
 import org.opengroup.osdu.indexer.service.SchemaEventsProcessor;
-import org.opengroup.osdu.indexer.service.SchemaService;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.context.annotation.RequestScope;
 
 import javax.inject.Inject;
@@ -51,34 +53,29 @@ import java.util.List;
 @RequestScope
 public class RecordIndexerApi {
 
+    @Inject
+    private IRequestInfo requestInfo;
     @Inject
     private IndexerService indexerService;
     @Inject
     private ReindexService reIndexService;
     @Inject
-    private SchemaService schemaService;
-    @Inject
     private SchemaEventsProcessor eventsProcessingService;
 
     // THIS IS AN INTERNAL USE API ONLY
     // THAT MEANS WE DON'T DOCUMENT IT IN SWAGGER, ACCESS IS LIMITED TO CLOUD TASK QUEUE CALLS ONLY
     @PostMapping(path = "/index-worker", consumes = "application/json")
     @Operation(hidden = true, summary = "", description = "")
-    public ResponseEntity<JobStatus> indexWorker (
+    public ResponseEntity<JobStatus> indexWorker(
              @NotNull(message = SwaggerDoc.REQUEST_VALIDATION_NOT_NULL_BODY)
              @Valid @RequestBody RecordChangedMessages recordChangedMessages) throws Exception {
 
-        if (recordChangedMessages.missingAccountId()) {
-            throw new AppException(org.apache.http.HttpStatus.SC_BAD_REQUEST, "Invalid tenant",
-                        String.format("Required header: '%s' not found", DpsHeaders.DATA_PARTITION_ID));
-        }
-        try {
-            if (recordChangedMessages == null) {
-                log.info("record change messages is null");
-            }
+        populateCorrelationIdIfExist(recordChangedMessages);
+
+        verifyDataPartitionId(recordChangedMessages);
 
-            Type listType = new TypeToken<List<RecordInfo>>() {
-            }.getType();
+        try {
+            Type listType = new TypeToken<List<RecordInfo>>() {}.getType();
             List<RecordInfo> recordInfos = new Gson().fromJson(recordChangedMessages.getData(), listType);
 
             if (recordInfos.size() == 0) {
@@ -96,6 +93,19 @@ public class RecordIndexerApi {
         }
     }
 
+    private void verifyDataPartitionId(RecordChangedMessages recordChangedMessages) {
+        if (recordChangedMessages.missingAccountId()) {
+            throw new AppException(org.apache.http.HttpStatus.SC_BAD_REQUEST, "Invalid tenant",
+                        String.format("Required header: '%s' not found", DpsHeaders.DATA_PARTITION_ID));
+        }
+    }
+
+    private void populateCorrelationIdIfExist(RecordChangedMessages recordChangedMessages) {
+        if (recordChangedMessages.hasCorrelationId()) {
+            this.requestInfo.getHeaders().put(DpsHeaders.CORRELATION_ID, recordChangedMessages.getCorrelationId());
+        }
+    }
+
     // THIS IS AN INTERNAL USE API ONLY
     // THAT MEANS WE DON'T DOCUMENT IT IN SWAGGER, ACCESS IS LIMITED TO CLOUD TASK QUEUE CALLS ONLY
     @PostMapping("/reindex-worker")

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

@@ -129,7 +129,7 @@ public class IndexerServiceImpl implements IndexerService {
 
         try {
             auditLogger.indexStarted(recordInfos.stream()
-                    .map(RecordInfo::getKind)
+                    .map(entry -> String.format("id=%s kind=%s operationType=%s", entry.getId(), entry.getKind(), entry.getOp()))
                     .collect(Collectors.toList()));
 
             // get upsert records

indexer-core/src/main/java/org/opengroup/osdu/indexer/util/BooleanFeatureFlagClient.java

@@ -55,7 +55,11 @@ public class BooleanFeatureFlagClient {
             PartitionInfo partitionInfo = partitionProvider.get(dataPartitionId);
             return partitionInfo;
         } catch (PartitionException e) {
-            logger.error(String.format("Error getting partition info for data-partition: %s", dataPartitionId), e);
+            if (e.getResponse() != null) {
+                logger.error(String.format("Error getting partition info for data-partition: %s. Message: %s. ResponseCode: %s.", dataPartitionId, e.getResponse().getBody(), e.getResponse().getResponseCode()), e);
+            } else {
+                logger.error(String.format("Error getting partition info for data-partition: %s.", dataPartitionId), e);
+            }
             throw e;
         }
     }

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

@@ -17,23 +17,26 @@ package org.opengroup.osdu.indexer.api;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 import static org.mockito.MockitoAnnotations.initMocks;
-
+import static org.mockito.Mockito.when;
 import com.google.gson.Gson;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 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.indexer.SchemaChangedMessages;
 import org.opengroup.osdu.core.common.model.search.RecordChangedMessages;
+import org.opengroup.osdu.core.common.provider.interfaces.IRequestInfo;
 import org.opengroup.osdu.indexer.service.IndexerService;
 import org.opengroup.osdu.indexer.service.SchemaEventsProcessor;
 import org.opengroup.osdu.indexer.service.SchemaService;
 import org.mockito.junit.MockitoJUnitRunner;
+import org.opengroup.osdu.indexer.util.IndexerQueueTaskBuilder;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 
@@ -56,7 +59,7 @@ public class RecordIndexerApiTest {
     @Mock
     private IndexerService indexService;
     @Mock
-    private SchemaService schemaService;
+    private IRequestInfo requestInfo;
     @Mock
     private SchemaEventsProcessor eventsProcessingService;
 
@@ -69,6 +72,7 @@ public class RecordIndexerApiTest {
 
         dpsHeaders.put(DpsHeaders.ACCOUNT_ID, this.ACCOUNT_ID);
         dpsHeaders.put(DpsHeaders.DATA_PARTITION_ID, this.DATA_PARTITION_ID);
+        when(this.requestInfo.getHeaders()).thenReturn(dpsHeaders);
     }
 
     @Test
@@ -86,6 +90,12 @@ public class RecordIndexerApiTest {
         should_return400_indexerWorkerTest(messageEmpty, String.format("Required header: '%s' not found", DpsHeaders.DATA_PARTITION_ID));
     }
 
+    @Test
+    public void should_addCorrelationIdToHeader_IfExists_indexWorkerTest() throws Exception {
+        this.sut.indexWorker(createRecordChangedMessage(recordMessageValid));
+        Mockito.verify(this.requestInfo.getHeaders()).put("correlation-id", "b5a281bd-f59d-4db2-9939-b2d85036fc7e");
+    }
+
     @Test
     public void should_return400_given_incorrectJsonFormatMessage_indexWorkerTest() {
         should_return400_indexerWorkerTest(messageWithIncorrectJsonFormat, "Unable to parse request payload.");