Merge branch 'kamalika/master' into 'master'

BugFix: 8622 Indexer publishes message to stale service bus topic

See merge request !379

NOTICE

@@ -708,8 +708,21 @@ The following software have components provided under the terms of this license:
 - spring-security-oauth2-jose (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
 - spring-security-test (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
 - springdoc-openapi-common (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-common)
-- springdoc-openapi-ui (from https://github.com/springdoc/springdoc-openapi, https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-ui)
+- springdoc-openapi-ui (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-ui)
 - springdoc-openapi-webmvc-core (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-webmvc-core)
+- springfox-bean-validators (from https://github.com/springfox/springfox)
+- springfox-boot-starter (from https://github.com/springfox/springfox)
+- springfox-core (from https://github.com/springfox/springfox)
+- springfox-data-rest (from https://github.com/springfox/springfox)
+- springfox-oas (from https://github.com/springfox/springfox)
+- springfox-schema (from https://github.com/springfox/springfox)
+- springfox-spi (from https://github.com/springfox/springfox)
+- springfox-spring-web (from https://github.com/springfox/springfox)
+- springfox-spring-webflux (from https://github.com/springfox/springfox)
+- springfox-spring-webmvc (from https://github.com/springfox/springfox)
+- springfox-swagger-common (from https://github.com/springfox/springfox)
+- springfox-swagger-ui (from https://github.com/springfox/springfox)
+- springfox-swagger2 (from https://github.com/springfox/springfox)
 - swagger-annotations (from https://repo1.maven.org/maven2/io/swagger/core/v3/swagger-annotations, https://repo1.maven.org/maven2/io/swagger/swagger-annotations)
 - swagger-core (from https://repo1.maven.org/maven2/io/swagger/core/v3/swagger-core, https://repo1.maven.org/maven2/io/swagger/swagger-core)
 - swagger-jaxrs (from https://repo1.maven.org/maven2/io/swagger/swagger-jaxrs)

provider/indexer-azure/src/main/java/org/opengroup/osdu/indexer/azure/config/AzureBootstrapConfig.java

@@ -39,6 +39,9 @@ public class AzureBootstrapConfig {
     @Value("${azure.servicebus.reindex.topic-name}")
     private String serviceBusReindexTopicName;
 
+    @Value("${publish.indexing.progress}")
+    private boolean publishIndexingProgress;
+
     @Value("${ELASTIC_CACHE_EXPIRATION}")
     private Integer elasticCacheExpiration;
 
@@ -63,6 +66,12 @@ public class AzureBootstrapConfig {
         return serviceBusReindexTopicName;
     }
 
+    @Bean
+    @Named("PUBLISH_TO_INDEXING_PROGRESS_TOPIC")
+    public Boolean publishIndexingProgress() {
+        return publishIndexingProgress;
+    }
+
     @Bean
     @Named("ELASTIC_CACHE_EXPIRATION")
     public Integer getElasticCacheExpiration() {

provider/indexer-azure/src/main/java/org/opengroup/osdu/indexer/azure/publish/PublisherImpl.java

@@ -56,6 +56,9 @@ public class PublisherImpl implements IPublisher {
     @Named("SERVICE_BUS_TOPIC")
     private String serviceBusTopic;
 
+    @Inject
+    @Named("PUBLISH_TO_INDEXING_PROGRESS_TOPIC")
+    private boolean publishToIndexingProgressTopic;
 
     @Override
     public void publishStatusChangedTagsToTopic(DpsHeaders headers, JobStatus indexerBatchStatus) throws Exception {
@@ -73,8 +76,10 @@ public class PublisherImpl implements IPublisher {
         message.setContentType("application/json");
 
         try {
-            logger.debug("Indexer publishes message " + headers.getCorrelationId());
-            topicClientFactory.getClient(headers.getPartitionId(), serviceBusTopic).send(message);
+            if(publishToIndexingProgressTopic) {
+                logger.debug("Indexer publishes message " + headers.getCorrelationId());
+                topicClientFactory.getClient(headers.getPartitionId(), serviceBusTopic).send(message);
+            }
         }
         catch (Exception e)
         {

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

@@ -68,6 +68,7 @@ azure.cosmosdb.database=${cosmosdb_database}
 #AzureServiceBusconfiguration
 azure.servicebus.topic-name=${servicebus_topic_name}
 azure.servicebus.reindex.topic-name=${reindex_topic_name}
+publish.indexing.progress=false
 
 #Indexer-Queue-header
 indexer.queue.key=abcd

provider/indexer-azure/src/test/java/org/opengroup/osdu/indexer/azure/service/PublisherImplTest.java

+package org.opengroup.osdu.indexer.azure.service;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.indexer.JobStatus;
+import org.opengroup.osdu.indexer.azure.publish.PublisherImpl;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+
+@RunWith(PowerMockRunner.class)
+public class PublisherImplTest {
+    @InjectMocks
+    private PublisherImpl publisher = new PublisherImpl();
+    @Mock
+    private JaxRsDpsLog logger;
+    private DpsHeaders dpsHeaders;
+    private JobStatus jobStatus;
+
+    @Test
+    public void indexer_should_not_publish_message_to_stale_service_bus_topic() throws Exception {
+        // Arrange
+        this.dpsHeaders = new DpsHeaders();
+        this.jobStatus = new JobStatus();
+        Map<String, String> messageAttributes = new HashMap<>();
+        messageAttributes.put(DpsHeaders.ACCOUNT_ID,  dpsHeaders.getPartitionIdWithFallbackToAccountId());
+        messageAttributes.put(DpsHeaders.DATA_PARTITION_ID,  dpsHeaders.getPartitionIdWithFallbackToAccountId());
+        messageAttributes.put(DpsHeaders.CORRELATION_ID, dpsHeaders.getCorrelationId());
+
+        // Act
+        publisher.publishStatusChangedTagsToTopic(dpsHeaders, jobStatus);
+
+        // Assert
+        verify(this.logger,never()).debug("Indexer publishes message " + dpsHeaders.getCorrelationId());
+    }
+}