Rename integration agent to connector

Author: jochenchrist

README.md

@@ -11,25 +11,25 @@ Using the SDK, you can build Java applications to automate data platform operati
 - Notify downstream consumers when data contract tests have failed
 - Publish data product costs and usage data to Data Mesh Manager
 
-This SDK is designed as a foundation for building data platform integrations that run as long-running agents on customer's data platform, e.g., as containers running in a Kubernetes cluster or any other container-runtime. 
+This SDK is designed as a foundation for building data platform integrations that run as long-running connectors on customer's data platform, e.g., as containers running in a Kubernetes cluster or any other container-runtime. 
 
 It interacts with the Data Mesh Manager APIs to send metadata and to subscribe to events to trigger actions in the data platform or with other services.
 
 
-Existing Integrations
+Existing Connectors
 ---
 
-We provide some agents for commonly-used platforms that that use this SDK and that can be used out-of-the-box or as a template for custom integrations:
+We provide some connectors for commonly-used platforms that that use this SDK and that can be used out-of-the-box or as a template for custom integrations:
 
-| Platform    | Integration                                                                                                | Synchronize Assets | Access Management | Remarks                 |
-|-------------|------------------------------------------------------------------------------------------------------------|--------------------|-------------------|-------------------------|
-| Databricks  | [datamesh-manager-agent-databricks](https://github.com/datamesh-manager/datamesh-manager-agent-databricks) | ✅                  | ✅                 | Uses Unity Catalog APIs |
-| Snowflake   |  [datamesh-manager-agent-snowflake](https://github.com/datamesh-manager/datamesh-manager-agent-snowflake)                                                                                                          | ✅                   | ✅                  |  Uses the Snowflake REST API           | 
-| AWS         |                                                                                                            |                    |                   | Coming soon             |
-| Google Cloud Platform | [datamesh-manager-agent-gcp](https://github.com/datamesh-manager/datamesh-manager-agent-gcp)                                                                                                           | ✅                    |✅                   | Uses BigQuery APIs             |
-| Azure       |                                                                                                            |                    |                   | Coming soon             |
-| datahub     |                                                                                                            |                    |                   | Coming soon             |
-| Collibra    |                                                                                                            |                    |                   | Coming soon             |
+| Platform              | Connector                                                                                                  | Synchronize Assets | Access Management | Remarks                     |
+|-----------------------|------------------------------------------------------------------------------------------------------------|--------------------|-------------------|-----------------------------|
+| Databricks            | [datamesh-manager-agent-databricks](https://github.com/datamesh-manager/datamesh-manager-agent-databricks) | ✅                  | ✅                 | Uses Unity Catalog APIs     |
+| Snowflake             | [datamesh-manager-agent-snowflake](https://github.com/datamesh-manager/datamesh-manager-agent-snowflake)   | ✅                  | ✅                 | Uses the Snowflake REST API | 
+| AWS                   |                                                                                                            |                    |                   | Coming soon                 |
+| Google Cloud Platform | [datamesh-manager-agent-gcp](https://github.com/datamesh-manager/datamesh-manager-agent-gcp)               | ✅                  | ✅                 | Uses BigQuery APIs          |
+| Azure                 |                                                                                                            |                    |                   | Coming soon                 |
+| datahub               |                                                                                                            |                    |                   | Coming soon                 |
+| Collibra              |                                                                                                            |                    |                   | Coming soon                 |
 
 If you are interested in further integration, please [contact us](https://entropy-data.atlassian.net/servicedesk/customer/portals).
 
@@ -90,10 +90,10 @@ public class MyAssetsProvider implements DataMeshManagerAssetsProvider {
 With this implementation, you can start an `DataMeshManagerAssetsSynchronizer`:
 
 ```java
-var agentid = "my-unique-assets-synchronization-agent-id";
+var connectorid = "my-unique-assets-synchronization-connector-id";
 var assetsProvider = new MyAssetsProvider();
-var assetsSynchronizer = new DataMeshManagerAssetsSynchronizer(agentid, client, assetsSupplier);
-assetsSynchronizer.start(); // This will start a long-running agent that calls the fetchAssets method periodically
+var assetsSynchronizer = new DataMeshManagerAssetsSynchronizer(connectorid, client, assetsSupplier);
+assetsSynchronizer.start(); // This will start a long-running connector that calls the fetchAssets method periodically
 ```
 
 ### Implement an EventListener (optional)
@@ -121,19 +121,19 @@ You can listen to any event from Data Mesh Manager. The SDK provides a method fo
 With this implementation, you can start an `DataMeshManagerEventListener`:
 
 ```java
-var agentid = "my-unique-event-listener-agent-id";
+var connectorid = "my-unique-event-listener-connector-id";
 var eventHandler = new MyEventHandler();
 var stateRepository = ... // see below
-var eventListener = new DataMeshManagerEventListener(agentid, client, eventHandler, stateRepository);
-eventListener.start(); // This will start a long-running agent that listens to events from Data Mesh Manager
+var eventListener = new DataMeshManagerEventListener(connectorid, client, eventHandler, stateRepository);
+eventListener.start(); // This will start a long-running connector that listens to events from Data Mesh Manager
 ```
 
-If you have multiple agents in an application, make sure to start the `start()` methods in separate threads.
+If you have multiple connectors in an application, make sure to start the `start()` methods in separate threads.
 
 ### State Repository
 
 The `DataMeshManagerEventListener` requires a `DataMeshManagerStateRepository` to store the `lastEventId` that has been processed.
-Also, you can use the state repository in other agents, if you need to store information what has been processed or what is the current state of your agent.
+Also, you can use the state repository in other connectors, if you need to store information what has been processed or what is the current state of your connector.
 You can implement this interface to store the state in a database, a file, or any other storage:
 
 ```java
@@ -146,8 +146,8 @@ public interface DataMeshManagerStateRepository {
 For your convenience, you can use the `DataMeshManagerStateRepositoryRemote` to store the state directly in the Data Mesh Manager:
 
 ```java
-var agentId = "my-unique-event-listener-agent-id";
-var stateRepository = new DataMeshManagerStateRepositoryRemote(agentId, client);
+var connectorId = "my-unique-event-listener-connector-id";
+var stateRepository = new DataMeshManagerStateRepositoryRemote(connectorId, client);
 ```
 
 and for testing there is also a `DataMeshManagerStateRepositoryInMemory`.

src/main/java/datameshmanager/sdk/DataMeshManagerAgentRegistration.java

@@ -13,30 +13,30 @@ public class DataMeshManagerAssetsSynchronizer {
 
   private static final Logger log = LoggerFactory.getLogger(DataMeshManagerAssetsSynchronizer.class);
 
-  private final String agentId;
+  private final String connectorId;
   private final DataMeshManagerClient client;
-  private final DataMeshManagerAgentRegistration agentRegistration;
+  private final DataMeshManagerConnectorRegistration connectorRegistration;
   private final DataMeshManagerAssetsProvider assetsProvider;
   private volatile boolean stopped = false;
 
   private Duration delay = Duration.parse("PT60M");
 
   public DataMeshManagerAssetsSynchronizer(
-      String agentId,
+      String connectorId,
       DataMeshManagerClient client,
       DataMeshManagerAssetsProvider assetsProvider) {
-    this.agentId = agentId;
+    this.connectorId = connectorId;
     this.client = client;
     this.assetsProvider = assetsProvider;
-    this.agentRegistration = new DataMeshManagerAgentRegistration(client, agentId, "assets-synchronizer");
+    this.connectorRegistration = new DataMeshManagerConnectorRegistration(client, connectorId, "assets-synchronizer");
 
-    this.agentRegistration.register();
+    this.connectorRegistration.register();
   }
 
   public void start() {
-    log.info("{}: start syncing assets", agentId);
+    log.info("{}: start syncing assets", connectorId);
 
-    // TODO error handling for agentRegistration
+    // TODO error handling for connectorRegistration
     // TODO error handling during while loop
 
     while (!this.stopped) {
@@ -52,11 +52,11 @@ public void start() {
 
   public void stop() {
     if (this.stopped) {
-      log.info("{}: Already stopped asset synchronization", agentId);
+      log.info("{}: Already stopped asset synchronization", connectorId);
       return;
     }
     this.stopped = true;
-    log.info("{}: Stopped syncing assets", agentId);
+    log.info("{}: Stopped syncing assets", connectorId);
   }
 
   public void synchronizeAssets() {

src/main/java/datameshmanager/sdk/DataMeshManagerAssetsSynchronizer.java

@@ -3,12 +3,12 @@
 import datameshmanager.sdk.client.ApiClient;
 import datameshmanager.sdk.client.api.AccessApi;
 import datameshmanager.sdk.client.api.AssetsApi;
+import datameshmanager.sdk.client.api.ConnectorsApi;
 import datameshmanager.sdk.client.api.CostsApi;
 import datameshmanager.sdk.client.api.DataContractsApi;
 import datameshmanager.sdk.client.api.DataProductsApi;
 import datameshmanager.sdk.client.api.DefinitionsApi;
 import datameshmanager.sdk.client.api.EventsApi;
-import datameshmanager.sdk.client.api.IntegrationsApi;
 import datameshmanager.sdk.client.api.SourceSystemsApi;
 import datameshmanager.sdk.client.api.TagsApi;
 import datameshmanager.sdk.client.api.TeamsApi;
@@ -29,7 +29,7 @@ public class DataMeshManagerClient {
   private final TagsApi tagsApi;
   private final TeamsApi teamsApi;
   private final TestResultsApi testResultsApi;
-  private final IntegrationsApi integrationsApi;
+  private final ConnectorsApi connectorsApi;
 
   public DataMeshManagerClient(String host, String apiKey) {
     var apiClient = new ApiClient();
@@ -48,7 +48,7 @@ public DataMeshManagerClient(String host, String apiKey) {
     this.tagsApi = new TagsApi(apiClient);
     this.teamsApi = new TeamsApi(apiClient);
     this.testResultsApi = new TestResultsApi(apiClient);
-    this.integrationsApi = new IntegrationsApi(apiClient);
+    this.connectorsApi = new ConnectorsApi(apiClient);
   }
 
   public ApiClient getApiClient() {
@@ -99,7 +99,7 @@ public TestResultsApi getTestResultsApi() {
     return testResultsApi;
   }
 
-  public IntegrationsApi getIntegrationsApi() {
-    return integrationsApi;
+  public ConnectorsApi getConnectorsApi() {
+    return connectorsApi;
   }
 }

src/main/java/datameshmanager/sdk/DataMeshManagerClient.java

@@ -0,0 +1,71 @@
+package datameshmanager.sdk;
+
+import datameshmanager.sdk.client.ApiException;
+import datameshmanager.sdk.client.model.Connector;
+import datameshmanager.sdk.client.model.ConnectorInfo;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DataMeshManagerConnectorRegistration {
+
+  private static final Logger log = LoggerFactory.getLogger(DataMeshManagerConnectorRegistration.class);
+
+  private final DataMeshManagerClient client;
+
+  /**
+   * The unique identifier of the connector.
+   * This is used to identify the connector in the Data Mesh Manager and to exchange state.
+   * The identifier should be constant across restarts of the connector.
+   */
+  private final String id;
+
+  /**
+   * The type of the connector (e.g. databricks-assets, databricks-access, aws-costs, ...).
+   */
+  private final String type;
+
+  public DataMeshManagerConnectorRegistration(DataMeshManagerClient client, String connectorId, String type) {
+    this.client = client;
+    this.id = Objects.requireNonNull(connectorId, "Connector ID is required");
+    this.type = Objects.requireNonNull(type, "Connector type is required");
+  }
+
+  public void register() {
+
+    Connector connector;
+    try {
+      log.debug("Checking if integration connector {} already exists", id);
+      connector = client.getConnectorsApi().getConnector(id);
+    } catch (ApiException e) {
+      if (e.getCode() == 404) {
+        connector = new Connector();
+      } else {
+        throw e;
+      }
+    }
+
+    connector
+        .id(id)
+        .info(new ConnectorInfo()
+            .type(type)
+        );
+
+    log.info("Registering integration connector {}", id);
+    client.getConnectorsApi().putConnector(id, connector);
+  }
+
+  public void delete() {
+    log.info("Deleting integration connector {}", this.id);
+    try {
+      client.getConnectorsApi().deleteConnector(this.id);
+    } catch (ApiException e) {
+      if (e.getCode() == 404) {
+        log.error("Integration connector with id {} already deleted", this.id);
+      } else {
+        throw e;
+      }
+    }
+  }
+
+}

src/main/java/datameshmanager/sdk/DataMeshManagerConnectorRegistration.java

@@ -52,39 +52,39 @@ public class DataMeshManagerEventListener {
 
   private static final Logger log = LoggerFactory.getLogger(DataMeshManagerEventListener.class);
 
-  private final String agentId;
+  private final String connectorId;
   private final DataMeshManagerEventHandler eventHandler;
   private final DataMeshManagerClient client;
   private final DataMeshManagerStateRepository stateRepository;
 
   private final ObjectMapper objectMapper;
-  private final DataMeshManagerAgentRegistration agentRegistration;
+  private final DataMeshManagerConnectorRegistration connectorRegistration;
 
   private boolean stopped = false;
   private Duration pollInterval = Duration.ofSeconds(5);
 
-  public DataMeshManagerEventListener(String agentId, String type, DataMeshManagerClient client, DataMeshManagerEventHandler eventHandler,
+  public DataMeshManagerEventListener(String connectorId, String type, DataMeshManagerClient client, DataMeshManagerEventHandler eventHandler,
       DataMeshManagerStateRepository stateRepository) {
-    this.agentId = Objects.requireNonNull(agentId, "agentId must not be null");
+    this.connectorId = Objects.requireNonNull(connectorId, "connectorId must not be null");
     this.eventHandler = Objects.requireNonNull(eventHandler, "eventHandler must not be null");
     this.client = Objects.requireNonNull(client, "client must not be null");
     this.stateRepository = Objects.requireNonNull(stateRepository, "stateRepository must not be null");
-    this.agentRegistration = new DataMeshManagerAgentRegistration(client, agentId, type);
+    this.connectorRegistration = new DataMeshManagerConnectorRegistration(client, connectorId, type);
 
     this.objectMapper = new ObjectMapper()
         .findAndRegisterModules()
         .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
 
-    this.agentRegistration.register();
+    this.connectorRegistration.register();
   }
 
   /**
    * Starts the event listener to poll for events from the DataMeshManager in an infinite loop.
    */
   public void start() {
-    log.info("{}: Start polling for events", agentId);
+    log.info("{}: Start polling for events", connectorId);
 
-    // TODO error handling for agentRegistration
+    // TODO error handling for connectorRegistration
 
     var lastEventId = getLastEventId();
     while (!this.stopped) {