Merge pull request #7138 from marius-patrascu/send-events-async

sendEvents Async
odpi · Jan 24, 2023 · 9f17b0f · 9f17b0f
2 parents ef42f73 + 9a8d504
commit 9f17b0f
Show file tree

Hide file tree

Showing 19 changed files with 297 additions and 88 deletions.
diff --git a/...ccessservices/assetconsumer/connectors/outtopic/AssetConsumerOutTopicServerConnector.java b/...ccessservices/assetconsumer/connectors/outtopic/AssetConsumerOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * AssetConsumerOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(AssetConsumerOutTopicEvent event) throws InvalidParameterE
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(AssetConsumerOutTopicEvent event) throws InvalidParameterE
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/.../accessservices/assetmanager/connectors/outtopic/AssetManagerOutTopicServerConnector.java b/.../accessservices/assetmanager/connectors/outtopic/AssetManagerOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * AssetManagerOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(AssetManagerOutTopicEvent event) throws InvalidParameterEx
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(AssetManagerOutTopicEvent event) throws InvalidParameterEx
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...data/accessservices/assetowner/connectors/outtopic/AssetOwnerOutTopicServerConnector.java b/...data/accessservices/assetowner/connectors/outtopic/AssetOwnerOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * AssetOwnerOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(AssetOwnerOutTopicEvent event) throws InvalidParameterExce
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(AssetOwnerOutTopicEvent event) throws InvalidParameterExce
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...ervices/communityprofile/connectors/outtopic/CommunityProfileOutTopicServerConnector.java b/...ervices/communityprofile/connectors/outtopic/CommunityProfileOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * CommunityProfileOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(CommunityProfileOutboundEvent event) throws InvalidParamet
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(CommunityProfileOutboundEvent event) throws InvalidParamet
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/.../odpi/openmetadata/accessservices/dataengine/server/handlers/DataEngineCommonHandler.java b/.../odpi/openmetadata/accessservices/dataengine/server/handlers/DataEngineCommonHandler.java
@@ -28,7 +28,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Date;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashSet;
@@ -209,11 +208,10 @@ protected void upsertExternalRelationship(String userId, String firstGUID, Strin
 
             TypeDef relationshipTypeDef = repositoryHelper.getTypeDefByName(userId, relationshipTypeName);
 
-            genericHandler.linkElementToElement(userId, externalSourceGUID, externalSourceName, firstGUID,
+            genericHandler.uncheckedLinkElementToElement(userId, externalSourceGUID, externalSourceName, firstGUID,
                     CommonMapper.GUID_PROPERTY_NAME, firstEntityTypeName, secondGUID, CommonMapper.GUID_PROPERTY_NAME,
                     secondEntityTypeName, false, false, null,
-                    relationshipTypeDef.getGUID(), relationshipTypeName, relationshipProperties, null,
-                    null, getNow(), methodName);
+                    relationshipTypeDef.getGUID(), relationshipTypeName, relationshipProperties,  getNow(), methodName);
         } else {
             Relationship originalRelationship = relationship.get();
             String relationshipGUID = originalRelationship.getGUID();

diff --git a/...i/openmetadata/accessservices/dataengine/server/handlers/DataEngineCommonHandlerTest.java b/...i/openmetadata/accessservices/dataengine/server/handlers/DataEngineCommonHandlerTest.java
@@ -135,10 +135,10 @@ void upsertExternalRelationship() throws InvalidParameterException, PropertyServ
                 ENTITY_TYPE_NAME, EXTERNAL_SOURCE_DE_QUALIFIED_NAME, null);
 
         verify(invalidParameterHandler, times(1)).validateUserId(USER, methodName);
-        verify(genericHandler, times(1)).linkElementToElement(USER,  EXTERNAL_SOURCE_DE_GUID, EXTERNAL_SOURCE_DE_QUALIFIED_NAME, FIRST_GUID,
+        verify(genericHandler, times(1)).uncheckedLinkElementToElement(USER,  EXTERNAL_SOURCE_DE_GUID, EXTERNAL_SOURCE_DE_QUALIFIED_NAME, FIRST_GUID,
                 CommonMapper.GUID_PROPERTY_NAME, ENTITY_TYPE_NAME, SECOND_GUID, CommonMapper.GUID_PROPERTY_NAME,
                 ENTITY_TYPE_NAME, false, false, null,
-                RELATIONSHIP_TYPE_GUID, RELATIONSHIP_TYPE_NAME, null, null, null, null, methodName);
+                RELATIONSHIP_TYPE_GUID, RELATIONSHIP_TYPE_NAME, null, null, methodName);
     }
 
     @Test

diff --git a/...tadata/accessservices/dataengine/connectors/intopic/DataEngineInTopicClientConnector.java b/...tadata/accessservices/dataengine/connectors/intopic/DataEngineInTopicClientConnector.java
@@ -11,6 +11,7 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
 
 
 /**
@@ -36,7 +37,7 @@ public void sendEvent(DataEngineEventHeader event) throws InvalidParameterExcept
         {
 
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -46,6 +47,17 @@ public void sendEvent(DataEngineEventHeader event) throws InvalidParameterExcept
             }
 
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...ta/accessservices/datamanager/connectors/outtopic/DataManagerOutTopicServerConnector.java b/...ta/accessservices/datamanager/connectors/outtopic/DataManagerOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * DataManagerOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(DataManagerOutboundEvent event) throws InvalidParameterExc
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(DataManagerOutboundEvent event) throws InvalidParameterExc
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...s/digitalarchitecture/connectors/outtopic/DigitalArchitectureOutTopicServerConnector.java b/...s/digitalarchitecture/connectors/outtopic/DigitalArchitectureOutTopicServerConnector.java
@@ -12,6 +12,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * DigitalArchitectureOutTopicServerConnector is the java implementation of the
@@ -34,7 +36,7 @@ public void sendEvent(DigitalArchitectureOutTopicEvent event) throws InvalidPara
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -43,6 +45,17 @@ public void sendEvent(DigitalArchitectureOutTopicEvent event) throws InvalidPara
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...ervices/governanceengine/connectors/outtopic/GovernanceEngineOutTopicServerConnector.java b/...ervices/governanceengine/connectors/outtopic/GovernanceEngineOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * GovernanceEngineOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(GovernanceEngineEvent event) throws InvalidParameterExcept
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(GovernanceEngineEvent event) throws InvalidParameterExcept
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...ervices/itinfrastructure/connectors/outtopic/ITInfrastructureOutTopicServerConnector.java b/...ervices/itinfrastructure/connectors/outtopic/ITInfrastructureOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * ITInfrastructureOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(ITInfrastructureOutTopicEvent event) throws InvalidParamet
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(ITInfrastructureOutTopicEvent event) throws InvalidParamet
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;

diff --git a/...sservices/securitymanager/connectors/outtopic/SecurityManagerOutTopicServerConnector.java b/...sservices/securitymanager/connectors/outtopic/SecurityManagerOutTopicServerConnector.java
@@ -11,6 +11,8 @@
 import org.odpi.openmetadata.frameworks.connectors.ffdc.InvalidParameterException;
 import org.odpi.openmetadata.repositoryservices.connectors.openmetadatatopic.OpenMetadataTopicSenderConnectorBase;
 
+import java.util.concurrent.CompletionException;
+
 
 /**
  * SecurityManagerOutTopicServerConnector is the java implementation of the
@@ -33,7 +35,7 @@ public void sendEvent(SecurityManagerOutTopicEvent event) throws InvalidParamete
         try
         {
             String eventString = objectMapper.writeValueAsString(event);
-            super.sendEvent(eventString);
+            super.sendEvent(eventString).join();
 
             if (super.auditLog != null)
             {
@@ -42,6 +44,17 @@ public void sendEvent(SecurityManagerOutTopicEvent event) throws InvalidParamete
                                           eventString);
             }
         }
+        catch (CompletionException error)
+        {
+            if (error.getCause() instanceof ConnectorCheckedException)
+            {
+                throw (ConnectorCheckedException) error.getCause();
+            }
+            else if (error.getCause() instanceof InvalidParameterException)
+            {
+                throw (InvalidParameterException) error.getCause();
+            }
+        }
         catch (InvalidParameterException | ConnectorCheckedException error)
         {
             throw error;