cp-adapter : persistance, horizontal scaling

Author: marcingajek-zf

edc-extensions/control-plane-adapter/README.md

@@ -4,23 +4,36 @@ The goal of this extension is to simplify the process of retrieving data out of
 
 Additional requirements, that affects the architecture of the extension:
 - can return data both in SYNC and ASYNC mode (currently only SYNC endpoint available)
-- can be persistent, so that process can be restored from the point where it was before application was stopped (not implemented yet)  
-- prepared to scale horizontally (not yet implemented)
+- can be persistent, so that process can be restored from the point where it was before application was stopped  
+- scaling horizontally (when persistence is added to configuration)
 - can retry failed part of the process (no need to start the process from the beginning)
 
-<b>Configuration:</b>
+<h2><u>Configuration:</u></h2>
 
-| Key                                              | Description                                                                                                                                                                                     | Mandatory | Default |
-|:-------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------|---------|
-| edc.cp.adapter.default.message.retry.number      | Number of retries of a message, in case of an error, within the internal process of retrieving DataReference                                                                                    | no   | 3       |
-| edc.cp.adapter.default.sync.request.timeout      | Timeout for synchronous request (in seconds), after witch 'timeout' error will be returned to the requesting client                                                                             | no   | 20      | 
-| edc.cp.adapter.messagebus.inmemory.thread.number | Number of threads running within the in-memory implementation of MessageBus                                                                                          _ _                        | no   | 10      |
-| edc.cp.adapter.reuse.contract.agreement          | Turn on/off reusing of existing contract agreements for the specific asset. Once the contract is agreed, the second request for the same asset will reuse the agreement. Value 1 = on, 0 = off. | no   | 1       |
-| edc.cp.adapter.cache.catalog.expire.after        | Number of seconds, after witch prevoiusly requested catalog will not be reused, and will be removed from catalog cache                                                                          | no   | 300     |
+| Key                                              | Description                                                                                                                                                                                                                            | Mandatory | Default |
+|:-------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------|---------|
+| edc.cp.adapter.default.message.retry.number      | Number of retries of a message, in case of an error, within the internal process of retrieving DataReference                                                                                                                           | no   | 3       |
+| edc.cp.adapter.default.sync.request.timeout      | Timeout for synchronous request (in seconds), after witch 'timeout' error will be returned to the requesting client                                                                                                                    | no   | 20      | 
+| edc.cp.adapter.messagebus.inmemory.thread.number | Number of threads running within the in-memory implementation of MessageBus                                                                                          _ _                                                               | no   | 10      |
+| edc.cp.adapter.reuse.contract.agreement          | Turn on/off reusing of existing contract agreements for the specific asset. Once the contract is agreed, the second request for the same asset will reuse the agreement (if exists) pulled from the EDC. Value 1 = on, 0 = off. | no   | 1       |
+| edc.cp.adapter.cache.catalog.expire.after        | Number of seconds, after witch prevoiusly requested catalog will not be reused, and will be removed from catalog cache                                                                                                                 | no   | 300     |
 | edc.cp.adapter.catalog.request.limit             | Maximum number of items taken from Catalog within single request. Requests are repeated until all offers of the query are retrieved                                                             | no   | 100     |
 
+By default, the extension works in "IN MEMORY" mode. This setup has some limitations:
++ It can work only within single EDC instance. If CP-adapter requests are handled by more than one EDC, data flow may be broken.
++ If the EDC instance is restarted, all running processes are lost.
 
-<b>How to use it:</b>
+To run CP-Adapter in "PERSISTENT" mode, You need to create a proper tables with [this](docs/schema.sql) script, and add the following configuration values to Your control-plane EDC properties file:
+
+| Key                               | Description |
+|-----------------------------------|-------------|
+| edc.datasource.cpadapter.name     | data source name |
+| edc.datasource.cpadapter.url      | data source url |
+| edc.datasource.cpadapter.user     | data source user |
+| edc.datasource.cpadapter.password | data source password |
+
+
+<h2><u>How to use it:</u></h2>
 1. Client sends a GET request with two parameters: assetId and the url of the provider control-plane:
 
    ```
@@ -33,7 +46,7 @@ Additional requirements, that affects the architecture of the extension:
    http://localhost:9193/api/v1/data/adapter/asset/sync/123?providerUrl=http://localhost:8182/api/v1/ids/data
    ```
 
-   Oprional request parameters:
+   Optional request parameters, that overwrite the settings for a single request:
 
    | Name | Description                                                                                                                                                                                                             |
    |--- |--- |

edc-extensions/control-plane-adapter/pom.xml

@@ -122,7 +122,7 @@
         <dependency>
             <groupId>org.eclipse.edc</groupId>
             <artifactId>sql-lease</artifactId>
-            <version>0.0.1-20221201-SNAPSHOT</version> <!-- TODO why needs version -->
+            <version>${org.eclipse.edc.version}</version>
         </dependency>
         <dependency>
             <groupId>org.eclipse.edc</groupId>

edc-extensions/control-plane-adapter/src/main/java/org/eclipse/tractusx/edc/cp/adapter/ApiAdapterConfig.java

@@ -23,10 +23,8 @@ public class ApiAdapterConfig {
       "edc.cp.adapter.default.sync.request.timeout";
   private static final String CATALOG_EXPIRE_AFTER_TIME =
       "edc.cp.adapter.cache.catalog.expire.after";
-  private static final String REUSE_CONTRACT_AGREEMENT =
-      "edc.cp.adapter.reuse.contract.agreement";
-  private static final String CATALOG_REQUEST_LIMIT =
-      "edc.cp.adapter.catalog.request.limit";
+  private static final String REUSE_CONTRACT_AGREEMENT = "edc.cp.adapter.reuse.contract.agreement";
+  private static final String CATALOG_REQUEST_LIMIT = "edc.cp.adapter.catalog.request.limit";
 
   private static final String DATASOURCE_NAME = "edc.datasource.cpadapter.name";
   private static final String DATASOURCE_URL = "edc.datasource.cpadapter.url";

edc-extensions/control-plane-adapter/src/test/java/org/eclipse/tractusx/edc/cp/adapter/messaging/SqlMessageBusTest.java

@@ -58,11 +58,11 @@ public void send_shouldCallListenerOnce() throws InterruptedException {
 
     // when
     messageBus.send(Channel.INITIAL, message);
-    Thread.sleep(50);
+    Thread.sleep(60);
     messageBus.deliverMessages(10);
 
     // then
-    Thread.sleep(50);
+    Thread.sleep(60);
     verify(listener, times(1)).process(any(DataReferenceRetrievalDto.class));
   }
 
@@ -78,6 +78,7 @@ public void send_shouldCallListenerWithRetryOnException() throws InterruptedExce
     // when
     messageBus.send(Channel.INITIAL, message);
     messageBus.deliverMessages(10);
+    Thread.sleep(60);
 
     // then
     verify(listener, times(2)).process(any(DataReferenceRetrievalDto.class));
@@ -95,7 +96,7 @@ public void send_shouldSendToDlqIfErrorLimitReached() throws InterruptedExceptio
 
     // when
     messageBus.send(Channel.INITIAL, message);
-    Thread.sleep(50);
+    Thread.sleep(60);
 
     // then
     verify(listenerService).getListener(eq(Channel.DLQ));
@@ -226,6 +227,11 @@ public void execute(TransactionBlock transactionBlock) {}
       public <T> T execute(ResultTransactionBlock<T> resultTransactionBlock) {
         return null;
       }
+
+      @Override
+      public void registerSynchronization(TransactionSynchronization transactionSynchronization) {
+
+      }
     };
   }
 }