Merge "Support for Patch across multiple data nodes"

4 files changed, 82 insertions, 65 deletions

diff --git a/cps-ri/src/main/java/org/onap/cps/spi/entities/FragmentEntityArranger.java b/cps-ri/src/main/java/org/onap/cps/spi/entities/FragmentEntityArranger.java
index 49e2dd2530..b7ce98e1ac 100644
--- a/cps-ri/src/main/java/org/onap/cps/spi/entities/FragmentEntityArranger.java
+++ b/cps-ri/src/main/java/org/onap/cps/spi/entities/FragmentEntityArranger.java
@@ -1,6 +1,7 @@
 /*
  *  ============LICENSE_START=======================================================
  *  Copyright (C) 2022 Nordix Foundation
+ *  Modifications Copyright (C) 2023 TechMahindra Ltd.
  *  ================================================================================
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -40,9 +41,11 @@ public class FragmentEntityArranger {
     public static Collection<FragmentEntity> toFragmentEntityTrees(final AnchorEntity anchorEntity,
                                                       final Collection<FragmentExtract> fragmentExtracts) {
         final Map<Long, FragmentEntity> fragmentEntityPerId = new HashMap<>();
-        for (final FragmentExtract fragmentExtract : fragmentExtracts) {
-            final FragmentEntity fragmentEntity = toFragmentEntity(anchorEntity, fragmentExtract);
-            fragmentEntityPerId.put(fragmentEntity.getId(), fragmentEntity);
+        if (fragmentExtracts !=  null) {
+            for (final FragmentExtract fragmentExtract : fragmentExtracts) {
+                final FragmentEntity fragmentEntity = toFragmentEntity(anchorEntity, fragmentExtract);
+                fragmentEntityPerId.put(fragmentEntity.getId(), fragmentEntity);
+            }
         }
         return reuniteChildrenWithTheirParents(fragmentEntityPerId);
     }
diff --git a/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java b/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java
index c26cd2fea8..3d2b87d7de 100644
--- a/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java
+++ b/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java
@@ -450,14 +450,25 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService
     }
 
     @Override
-    public void updateDataLeaves(final String dataspaceName, final String anchorName, final String xpath,
-                                 final Map<String, Serializable> updateLeaves) {
+    public void batchUpdateDataLeaves(final String dataspaceName, final String anchorName,
+                                        final Map<String, Map<String, Serializable>> updatedLeavesPerXPath) {
         final AnchorEntity anchorEntity = getAnchorEntity(dataspaceName, anchorName);
-        final FragmentEntity fragmentEntity = getFragmentEntity(anchorEntity, xpath);
-        final String currentLeavesAsString = fragmentEntity.getAttributes();
-        final String mergedLeaves = mergeLeaves(updateLeaves, currentLeavesAsString);
-        fragmentEntity.setAttributes(mergedLeaves);
-        fragmentRepository.save(fragmentEntity);
+
+        final Collection<String> xpathsOfUpdatedLeaves = updatedLeavesPerXPath.keySet();
+        final Collection<FragmentEntity> fragmentEntities = getFragmentEntities(anchorEntity, xpathsOfUpdatedLeaves,
+                FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS);
+
+        for (final FragmentEntity fragmentEntity : fragmentEntities) {
+            final Map<String, Serializable> updatedLeaves = updatedLeavesPerXPath.get(fragmentEntity.getXpath());
+            final String mergedLeaves = mergeLeaves(updatedLeaves, fragmentEntity.getAttributes());
+            fragmentEntity.setAttributes(mergedLeaves);
+        }
+
+        try {
+            fragmentRepository.saveAll(fragmentEntities);
+        } catch (final StaleStateException staleStateException) {
+            retryUpdateDataNodesIndividually(anchorEntity, fragmentEntities);
+        }
     }
 
     @Override
@@ -687,9 +698,13 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService
     }
 
     private String mergeLeaves(final Map<String, Serializable> updateLeaves, final String currentLeavesAsString) {
-        final Map<String, Serializable> currentLeavesAsMap = currentLeavesAsString.isEmpty()
-            ? new HashMap<>() : jsonObjectMapper.convertJsonString(currentLeavesAsString, Map.class);
-        currentLeavesAsMap.putAll(updateLeaves);
+        Map<String, Serializable> currentLeavesAsMap = new HashMap<>();
+        if (currentLeavesAsString != null) {
+            currentLeavesAsMap = currentLeavesAsString.isEmpty()
+                    ? new HashMap<>() : jsonObjectMapper.convertJsonString(currentLeavesAsString, Map.class);
+            currentLeavesAsMap.putAll(updateLeaves);
+        }
+
         if (currentLeavesAsMap.isEmpty()) {
             return "";
         }
diff --git a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy
index 67ccc805ae..080e348284 100755
--- a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy
+++ b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy
@@ -320,36 +320,6 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase {
     }
 
     @Sql([CLEAR_DATA, SET_DATA])
-    def 'Update data node leaves.'() {
-        when: 'update is performed for leaves'
-            objectUnderTest.updateDataLeaves(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES,
-                    '/parent-200/child-201', ['leaf-value': 'new'])
-        then: 'leaves are updated for selected data node'
-            def updatedFragment = fragmentRepository.getReferenceById(DATA_NODE_202_FRAGMENT_ID)
-            def updatedLeaves = getLeavesMap(updatedFragment)
-            assert updatedLeaves.size() == 1
-            assert updatedLeaves.'leaf-value' == 'new'
-        and: 'existing child entry remains as is'
-            def childFragment = updatedFragment.childFragments.iterator().next()
-            def childLeaves = getLeavesMap(childFragment)
-            assert childFragment.id == CHILD_OF_DATA_NODE_202_FRAGMENT_ID
-            assert childLeaves.'leaf-value' == 'original'
-    }
-
-    @Sql([CLEAR_DATA, SET_DATA])
-    def 'Update data leaves error scenario: #scenario.'() {
-        when: 'attempt to update data node for #scenario'
-            objectUnderTest.updateDataLeaves(dataspaceName, anchorName, xpath, ['leaf-name': 'leaf-value'])
-        then: 'a #expectedException is thrown'
-            thrown(expectedException)
-        where: 'the following data is used'
-            scenario                 | dataspaceName  | anchorName                        | xpath                 || expectedException
-            'non-existing dataspace' | 'NO DATASPACE' | 'not relevant'                    | '/not relevant'       || DataspaceNotFoundException
-            'non-existing anchor'    | DATASPACE_NAME | 'NO ANCHOR'                       | '/not relevant'       || AnchorNotFoundException
-            'non-existing xpath'     | DATASPACE_NAME | ANCHOR_FOR_DATA_NODES_WITH_LEAVES | '/NON-EXISTING-XPATH' || DataNodeNotFoundException
-    }
-
-    @Sql([CLEAR_DATA, SET_DATA])
     def 'Update data nodes and descendants by removing descendants.'() {
         given: 'data nodes with leaves updated, no children'
             def submittedDataNodes = [buildDataNode('/parent-200/child-201', ['leaf-value': 'new'], [])]
diff --git a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy
index 8a5838827a..e8921b3ed0 100644
--- a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy
+++ b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy
@@ -38,6 +38,7 @@ import org.onap.cps.spi.utils.SessionManager
 import org.onap.cps.utils.JsonObjectMapper
 import org.springframework.dao.DataIntegrityViolationException
 import spock.lang.Specification
+import java.util.stream.Collectors
 
 class CpsDataPersistenceServiceSpec extends Specification {
 
@@ -68,6 +69,53 @@ class CpsDataPersistenceServiceSpec extends Specification {
             2 * mockFragmentRepository.save(_)
     }
 
+    def 'Handling of StaleStateException (caused by concurrent updates) during patch operation for data nodes.'() {
+        given: 'the system can update one datanode and has two more datanodes that throw an exception while updating'
+            def dataNodes = createDataNodesAndMockRepositoryMethodSupportingThem([
+                    '/node1': 'OK',
+                    '/node2': 'EXCEPTION',
+                    '/node3': 'EXCEPTION'])
+            def updatedLeavesPerXPath = dataNodes.stream()
+                    .collect(Collectors.toMap(DataNode::getXpath, DataNode::getLeaves))
+        and: 'the batch update will therefore also fail'
+            mockFragmentRepository.saveAll(*_) >> { throw new StaleStateException("concurrent updates") }
+        when: 'attempt batch update data nodes'
+            objectUnderTest.batchUpdateDataLeaves('some-dataspace', 'some-anchor', updatedLeavesPerXPath)
+        then: 'concurrency exception is thrown'
+            def thrown = thrown(ConcurrencyException)
+            assert thrown.message == 'Concurrent Transactions'
+        and: 'it does not contain the successful datanode'
+            assert !thrown.details.contains('/node1')
+        and: 'it contains the failed datanodes'
+            assert thrown.details.contains('/node2')
+            assert thrown.details.contains('/node3')
+    }
+
+    def 'Batch update data node leaves and descendants: #scenario'(){
+        given: 'the fragment repository returns fragment entities related to the xpath inputs'
+            mockFragmentRepository.findExtractsWithDescendants(_, [] as Set, _) >> []
+            mockFragmentRepository.findExtractsWithDescendants(_, ['/test/xpath'] as Set, _) >> [
+                    mockFragmentExtract(1, null, 123, '/test/xpath', "{\"id\":\"testId1\"}")
+            ]
+            mockFragmentRepository.findExtractsWithDescendants(123, ['/test/xpath1', '/test/xpath2'] as Set, _) >> [
+                    mockFragmentExtract(1, null, 123, '/test/xpath1', "{\"id\":\"testId1\"}"),
+                    mockFragmentExtract(2, null, 123, '/test/xpath2', "{\"id\":\"testId1\"}")
+            ]
+        when: 'replace data node tree'
+            objectUnderTest.batchUpdateDataLeaves('dataspaceName', 'anchorName',
+                    dataNodes.stream().collect(Collectors.toMap(DataNode::getXpath, DataNode::getLeaves)))
+        then: 'call fragment repository save all method'
+            1 * mockFragmentRepository.saveAll({fragmentEntities ->
+                assert fragmentEntities as List == expectedFragmentEntities
+                assert fragmentEntities.size() == expectedSize
+            })
+        where: 'the following Data Type is passed'
+            scenario                         | dataNodes                                                                                                                              | expectedSize || expectedFragmentEntities
+            'empty data node list'           | []                                                                                                                                     | 0            || []
+            'one data node in list'          | [new DataNode(xpath: '/test/xpath', leaves: ['id': 'testId'])]                                                                         | 1            || [new FragmentEntity(xpath: '/test/xpath', attributes: '{"id":"testId"}', anchor: anchorEntity)]
+            'multiple data nodes'            | [new DataNode(xpath: '/test/xpath1', leaves: ['id': 'newTestId1']), new DataNode(xpath: '/test/xpath2', leaves: ['id': 'newTestId2'])] | 2            || [new FragmentEntity(xpath: '/test/xpath2', attributes: '{"id":"newTestId2"}', anchor: anchorEntity), new FragmentEntity(xpath: '/test/xpath1', attributes: '{"id":"newTestId1"}', anchor: anchorEntity)]
+    }
+
     def 'Handling of StaleStateException (caused by concurrent updates) during update data nodes and descendants.'() {
         given: 'the system can update one datanode and has two more datanodes that throw an exception while updating'
             def dataNodes = createDataNodesAndMockRepositoryMethodSupportingThem([
@@ -81,7 +129,7 @@ class CpsDataPersistenceServiceSpec extends Specification {
         then: 'concurrency exception is thrown'
             def thrown = thrown(ConcurrencyException)
             assert thrown.message == 'Concurrent Transactions'
-        and: 'it does not contain the successfull datanode'
+        and: 'it does not contain the successful datanode'
             assert !thrown.details.contains('/node1')
         and: 'it contains the failed datanodes'
             assert thrown.details.contains('/node2')
@@ -157,26 +205,7 @@ class CpsDataPersistenceServiceSpec extends Specification {
             1 * mockSessionManager.lockAnchor('mySessionId', 'myDataspaceName', 'myAnchorName', 123L)
     }
 
-    def 'update data node leaves: #scenario'(){
-        given: 'A node exists for the given xpath'
-            mockFragmentRepository.getByAnchorAndXpath(_, '/some/xpath') >> new FragmentEntity(xpath: '/some/xpath', attributes:  existingAttributes)
-        when: 'the node leaves are updated'
-            objectUnderTest.updateDataLeaves('some-dataspace', 'some-anchor', '/some/xpath', newAttributes as Map<String, Serializable>)
-        then: 'the fragment entity saved has the original and new attributes'
-            1 * mockFragmentRepository.save({fragmentEntity -> {
-                assert fragmentEntity.getXpath() == '/some/xpath'
-                assert fragmentEntity.getAttributes() == mergedAttributes
-            }})
-        where: 'the following attributes combinations are used'
-            scenario                      | existingAttributes     | newAttributes         | mergedAttributes
-            'add new leaf'                | '{"existing":"value"}' | ["new":"value"]       | '{"existing":"value","new":"value"}'
-            'update existing leaf'        | '{"existing":"value"}' | ["existing":"value2"] | '{"existing":"value2"}'
-            'update nothing with nothing' | ''                     | []                    | ''
-            'update with nothing'         | '{"existing":"value"}' | []                    | '{"existing":"value"}'
-            'update with same value'      | '{"existing":"value"}' | ["existing":"value"]  | '{"existing":"value"}'
-    }
-
-    def 'update data node and descendants: #scenario'(){
+    def 'Replace data node and descendants: #scenario'(){
         given: 'the fragment repository returns fragment entities related to the xpath inputs'
             mockFragmentRepository.findExtractsWithDescendants(_, [] as Set, _) >> []
             mockFragmentRepository.findExtractsWithDescendants(_, ['/test/xpath'] as Set, _) >> [
@@ -192,7 +221,7 @@ class CpsDataPersistenceServiceSpec extends Specification {
             'one data node in list'          | [new DataNode(xpath: '/test/xpath', leaves: ['id': 'testId'], childDataNodes: [])] || [new FragmentEntity(xpath: '/test/xpath', attributes: '{"id":"testId"}', anchor: anchorEntity, childFragments: [])]
     }
 
-    def 'update data nodes and descendants'() {
+    def 'Replace data nodes and descendants'() {
         given: 'the fragment repository returns fragment entities related to the xpath inputs'
             mockFragmentRepository.findExtractsWithDescendants(_, ['/test/xpath1', '/test/xpath2'] as Set, _) >> [
                 mockFragmentExtract(1, null, 123, '/test/xpath1', null),