Add validations for input/output ports (#25601)

* Add validations for input/output ports

* Add validations for input/output ports

* Fix tests

* addresss comments

---------

Co-authored-by: Sid <30566406+siddhant1@users.noreply.github.com>
Co-authored-by: Sriharsha Chintalapani <harsha.ch@gmail.com>

Author: 3 people

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DataProductDomainMigrationIT.java

@@ -169,7 +169,18 @@ void testDataProductDomainMigrationWithInputOutputPorts(TestNamespace ns) throws
     Table inputTable = createTestTableInDomain(ns, "input_port_table", sourceDomain);
     Table outputTable = createTestTableInDomain(ns, "output_port_table", sourceDomain);
 
-    // Add input port
+    // Output port assets must first be added as HAS assets (data product assets)
+    // before they can be designated as output ports
+    List<EntityReference> outputAssetRefs =
+        List.of(
+            new EntityReference()
+                .withId(outputTable.getId())
+                .withType("table")
+                .withFullyQualifiedName(outputTable.getFullyQualifiedName()));
+    BulkAssets outputAssetRequest = new BulkAssets().withAssets(outputAssetRefs);
+    client.dataProducts().bulkAddAssets(dataProduct.getFullyQualifiedName(), outputAssetRequest);
+
+    // Add input port (no HAS requirement - input ports can come from any domain)
     List<EntityReference> inputPorts =
         List.of(
             new EntityReference()
@@ -179,7 +190,7 @@ void testDataProductDomainMigrationWithInputOutputPorts(TestNamespace ns) throws
     BulkAssets inputRequest = new BulkAssets().withAssets(inputPorts);
     client.dataProducts().bulkAddInputPorts(dataProduct.getFullyQualifiedName(), inputRequest);
 
-    // Add output port
+    // Add output port (requires HAS relationship first)
     List<EntityReference> outputPorts =
         List.of(
             new EntityReference()

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DataProductResourceIT.java

@@ -1,15 +1,29 @@
 package org.openmetadata.sdk.exceptions;
 
 public class InvalidRequestException extends OpenMetadataException {
+  private final String responseBody;
+
   public InvalidRequestException(String message) {
     super(message, 400);
+    this.responseBody = null;
   }
 
   public InvalidRequestException(String message, Throwable cause) {
     super(message, cause, 400, null);
+    this.responseBody = null;
   }
 
   public InvalidRequestException(String message, String errorCode) {
     super(message, 400, errorCode);
+    this.responseBody = null;
+  }
+
+  public InvalidRequestException(String message, String errorCode, String responseBody) {
+    super(message, 400, errorCode);
+    this.responseBody = responseBody;
+  }
+
+  public String getResponseBody() {
+    return responseBody;
   }
 }

openmetadata-sdk/src/main/java/org/openmetadata/sdk/exceptions/InvalidRequestException.java

@@ -323,7 +323,7 @@ private void handleErrorResponse(Response response) throws OpenMetadataException
     // Throw appropriate exception based on status code
     switch (statusCode) {
       case 400:
-        throw new InvalidRequestException(errorMessage);
+        throw new InvalidRequestException(errorMessage, (String) null, responseBodyString);
       case 401:
         throw new AuthenticationException(errorMessage);
       case 409:

openmetadata-sdk/src/main/java/org/openmetadata/sdk/network/OpenMetadataHttpClient.java

@@ -1802,6 +1802,16 @@ List<UUID> findToIds(
         @Bind("relation") int relation,
         @Bind("toEntity") String toEntity);
 
+    @SqlQuery(
+        "SELECT COUNT(*) FROM entity_relationship "
+            + "WHERE fromId = :fromId AND toId = :toId AND fromEntity = :fromEntity AND toEntity = :toEntity AND relation = :relation")
+    int existsRelationship(
+        @BindUUID("fromId") UUID fromId,
+        @BindUUID("toId") UUID toId,
+        @Bind("fromEntity") String fromEntity,
+        @Bind("toEntity") String toEntity,
+        @Bind("relation") int relation);
+
     @SqlQuery(
         "SELECT fromId, COUNT(toId) FROM entity_relationship "
             + "WHERE fromId IN (<fromIds>) AND fromEntity = :fromEntity AND relation = :relation AND toEntity = :toEntity "

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java

@@ -35,10 +35,13 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.sqlobject.transaction.Transaction;
 import org.openmetadata.schema.EntityInterface;
@@ -287,16 +290,68 @@ private BulkOperationResult executeBulkPortsOperation(
     BulkOperationResult result =
         new BulkOperationResult().withStatus(ApiStatus.SUCCESS).withDryRun(false);
     List<BulkResponse> success = new ArrayList<>();
+    List<BulkResponse> failed = new ArrayList<>();
 
     List<EntityReference> assets = new ArrayList<>(listOrEmpty(request.getAssets()));
     EntityUtil.populateEntityReferences(assets);
 
     String fieldName = relationship == Relationship.INPUT_PORT ? "inputPorts" : "outputPorts";
 
+    Relationship oppositeRelationship =
+        relationship == Relationship.INPUT_PORT
+            ? Relationship.OUTPUT_PORT
+            : Relationship.INPUT_PORT;
+
+    Set<UUID> oppositePortIds = Set.of();
+    Set<UUID> dataProductAssetIds = Set.of();
+    if (isAdd) {
+      oppositePortIds =
+          daoCollection
+              .relationshipDAO()
+              .findTo(dataProduct.getId(), DATA_PRODUCT, oppositeRelationship.ordinal())
+              .stream()
+              .map(CollectionDAO.EntityRelationshipRecord::getId)
+              .collect(Collectors.toCollection(HashSet::new));
+      if (relationship == Relationship.OUTPUT_PORT) {
+        dataProductAssetIds =
+            daoCollection
+                .relationshipDAO()
+                .findTo(dataProduct.getId(), DATA_PRODUCT, Relationship.HAS.ordinal())
+                .stream()
+                .map(CollectionDAO.EntityRelationshipRecord::getId)
+                .collect(Collectors.toCollection(HashSet::new));
+      }
+    }
+
     for (EntityReference ref : assets) {
       result.setNumberOfRowsProcessed(result.getNumberOfRowsProcessed() + 1);
 
       if (isAdd) {
+        if (oppositePortIds.contains(ref.getId())) {
+          String oppositePortType =
+              oppositeRelationship == Relationship.INPUT_PORT ? "input" : "output";
+          String msg =
+              String.format(
+                  "Asset '%s' is already part of %s ports and cannot be added to %s",
+                  ref.getFullyQualifiedName(), oppositePortType, fieldName);
+          failed.add(new BulkResponse().withRequest(ref).withMessage(msg));
+          result.setNumberOfRowsFailed(result.getNumberOfRowsFailed() + 1);
+          result.setStatus(ApiStatus.PARTIAL_SUCCESS);
+          continue;
+        }
+
+        if (relationship == Relationship.OUTPUT_PORT
+            && !dataProductAssetIds.contains(ref.getId())) {
+          String msg =
+              String.format(
+                  "Asset '%s' must belong to the data product before it can be added as an output port",
+                  ref.getFullyQualifiedName());
+          failed.add(new BulkResponse().withRequest(ref).withMessage(msg));
+          result.setNumberOfRowsFailed(result.getNumberOfRowsFailed() + 1);
+          result.setStatus(ApiStatus.PARTIAL_SUCCESS);
+          continue;
+        }
+
         addRelationship(
             dataProduct.getId(), ref.getId(), DATA_PRODUCT, ref.getType(), relationship);
       } else {
@@ -311,11 +366,17 @@ private BulkOperationResult executeBulkPortsOperation(
           .onEntityUpdated(dataProduct.getEntityReference(), null);
     }
 
-    result.withSuccessRequest(success);
+    result.withSuccessRequest(success).withFailedRequest(failed);
 
-    if (result.getStatus().equals(ApiStatus.SUCCESS)) {
+    if (success.isEmpty() && !failed.isEmpty()) {
+      result.setStatus(ApiStatus.FAILURE);
+    }
+
+    if (!success.isEmpty()) {
+      List<EntityReference> successAssets =
+          success.stream().map(r -> (EntityReference) r.getRequest()).collect(Collectors.toList());
       ChangeDescription change =
-          addBulkAddRemoveChangeDescription(dataProduct.getVersion(), isAdd, assets, null);
+          addBulkAddRemoveChangeDescription(dataProduct.getVersion(), isAdd, successAssets, null);
       if (!change.getFieldsAdded().isEmpty()) {
         change.getFieldsAdded().get(0).setName(fieldName);
       }