Merge branch 'master' into remove-page-size

Author: sid-acryl

.github/workflows/dagster-plugin.yml

@@ -31,9 +31,9 @@ jobs:
       DATAHUB_TELEMETRY_ENABLED: false
     strategy:
       matrix:
-        python-version: ["3.8", "3.10"]
+        python-version: ["3.9", "3.10"]
         include:
-          - python-version: "3.8"
+          - python-version: "3.9"
             extraPythonRequirement: "dagster>=1.3.3"
           - python-version: "3.10"
             extraPythonRequirement: "dagster>=1.3.3"

build.gradle

@@ -34,6 +34,7 @@ buildscript {
   // Releases: https://github.com/linkedin/rest.li/blob/master/CHANGELOG.md
   ext.pegasusVersion = '29.57.0'
   ext.mavenVersion = '3.6.3'
+  ext.versionGradle = '8.11.1'
   ext.springVersion = '6.1.13'
   ext.springBootVersion = '3.2.9'
   ext.springKafkaVersion = '3.1.6'
@@ -78,7 +79,7 @@ buildscript {
 
 plugins {
   id 'com.gorylenko.gradle-git-properties' version '2.4.1'
-  id 'com.github.johnrengelman.shadow' version '8.1.1' apply false
+  id 'com.gradleup.shadow' version '8.3.5' apply false
   id 'com.palantir.docker' version '0.35.0' apply false
   id 'com.avast.gradle.docker-compose' version '0.17.6'
   id "com.diffplug.spotless" version "6.23.3"
@@ -499,3 +500,8 @@ subprojects {
     }
   }
 }
+
+wrapper {
+  gradleVersion = project.versionGradle
+  distributionType = Wrapper.DistributionType.ALL
+}

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/GmsGraphQLEngine.java

@@ -1318,7 +1318,8 @@ private void configureMutationResolvers(final RuntimeWiring.Builder builder) {
               .dataFetcher("updateQuery", new UpdateQueryResolver(this.queryService))
               .dataFetcher("deleteQuery", new DeleteQueryResolver(this.queryService))
               .dataFetcher(
-                  "createDataProduct", new CreateDataProductResolver(this.dataProductService))
+                  "createDataProduct",
+                  new CreateDataProductResolver(this.dataProductService, this.entityService))
               .dataFetcher(
                   "updateDataProduct", new UpdateDataProductResolver(this.dataProductService))
               .dataFetcher(

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/dataproduct/CreateDataProductResolver.java

@@ -10,8 +10,11 @@
 import com.linkedin.datahub.graphql.exception.AuthorizationException;
 import com.linkedin.datahub.graphql.generated.CreateDataProductInput;
 import com.linkedin.datahub.graphql.generated.DataProduct;
+import com.linkedin.datahub.graphql.generated.OwnerEntityType;
+import com.linkedin.datahub.graphql.resolvers.mutate.util.OwnerUtils;
 import com.linkedin.datahub.graphql.types.dataproduct.mappers.DataProductMapper;
 import com.linkedin.entity.EntityResponse;
+import com.linkedin.metadata.entity.EntityService;
 import com.linkedin.metadata.service.DataProductService;
 import graphql.schema.DataFetcher;
 import graphql.schema.DataFetchingEnvironment;
@@ -24,6 +27,7 @@
 public class CreateDataProductResolver implements DataFetcher<CompletableFuture<DataProduct>> {
 
   private final DataProductService _dataProductService;
+  private final EntityService _entityService;
 
   @Override
   public CompletableFuture<DataProduct> get(final DataFetchingEnvironment environment)
@@ -56,6 +60,8 @@ public CompletableFuture<DataProduct> get(final DataFetchingEnvironment environm
                 context.getOperationContext(),
                 dataProductUrn,
                 UrnUtils.getUrn(input.getDomainUrn()));
+            OwnerUtils.addCreatorAsOwner(
+                context, dataProductUrn.toString(), OwnerEntityType.CORP_USER, _entityService);
             EntityResponse response =
                 _dataProductService.getDataProductEntityResponse(
                     context.getOperationContext(), dataProductUrn);

docker/profiles/docker-compose.gms.yml

@@ -99,7 +99,6 @@ x-datahub-gms-service: &datahub-gms-service
     - ${DATAHUB_LOCAL_GMS_ENV:-empty2.env}
   environment: &datahub-gms-env
     <<: [*primary-datastore-mysql-env, *graph-datastore-search-env, *search-datastore-env, *datahub-quickstart-telemetry-env, *kafka-env]
-    ELASTICSEARCH_QUERY_CUSTOM_CONFIG_FILE: ${ELASTICSEARCH_QUERY_CUSTOM_CONFIG_FILE:-search_config.yaml}
     ALTERNATE_MCP_VALIDATION: ${ALTERNATE_MCP_VALIDATION:-true}
     STRICT_URN_VALIDATION_ENABLED: ${STRICT_URN_VALIDATION_ENABLED:-true}
   healthcheck:
@@ -126,7 +125,6 @@ x-datahub-gms-service-dev: &datahub-gms-service-dev
     - ${DATAHUB_LOCAL_GMS_ENV:-empty2.env}
   environment: &datahub-gms-dev-env
     <<: [*datahub-dev-telemetry-env, *datahub-gms-env]
-    ELASTICSEARCH_QUERY_CUSTOM_CONFIG_FILE: ${ELASTICSEARCH_QUERY_CUSTOM_CONFIG_FILE:-search_config.yaml}
     SKIP_ELASTICSEARCH_CHECK: false
     JAVA_TOOL_OPTIONS: '-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5001'
     BOOTSTRAP_SYSTEM_UPDATE_WAIT_FOR_SYSTEM_UPDATE: false

docs/advanced/mcp-mcl.md

@@ -218,3 +218,6 @@ Another form of conditional writes which considers the existence of an aspect or
 
 `CREATE_ENTITY` - Create the aspect if no aspects exist for the entity.
 
+By default, a validation exception is thrown if the `CREATE`/`CREATE_ENTITY` constraint is violated. If the write operation
+should be dropped without considering it an exception, then add the following header: `If-None-Match: *` to the MCP.
+

docs/plugins.md

@@ -65,14 +65,14 @@ The sample authenticator implementation can be found at [Authenticator Sample](.
 3. Use `getResourceAsStream` to read files: If your plugin read any configuration file like properties or YAML or JSON or xml then use `this.getClass().getClassLoader().getResourceAsStream("<file-name>")` to read that file from DataHub GMS plugin's class-path. For DataHub GMS resource look-up behavior please refer [Plugin Installation](#plugin-installation) section. Sample code of `getResourceAsStream` is available in sample Authenticator plugin [TestAuthenticator.java](../metadata-service/plugin/src/test/sample-test-plugins/src/main/java/com/datahub/plugins/test/TestAuthenticator.java).
 
 
-4. Bundle your Jar: Use `com.github.johnrengelman.shadow` gradle plugin to create an uber jar. 
+4. Bundle your Jar: Use `com.gradleup.shadow` gradle plugin to create an uber jar. 
 
    To see an example of building an uber jar, check out the `build.gradle` file for the apache-ranger-plugin file of [Apache Ranger Plugin](https://github.com/acryldata/datahub-ranger-auth-plugin/tree/main/apache-ranger-plugin) for reference. 
 
    Exclude signature files as shown in below `shadowJar` task.
 
    ```groovy
-     apply plugin: 'com.github.johnrengelman.shadow';
+     apply plugin: 'com.gradleup.shadow';
      shadowJar {
          // Exclude com.datahub.plugins package and files related to jar signature   
          exclude "META-INF/*.RSA", "META-INF/*.SF","META-INF/*.DSA"
@@ -152,14 +152,14 @@ The sample authorizer implementation can be found at [Authorizer Sample](https:/
 
 3. Use `getResourceAsStream` to read files: If your plugin read any configuration file like properties or YAML or JSON or xml then use `this.getClass().getClassLoader().getResourceAsStream("<file-name>")` to read that file from DataHub GMS plugin's class-path. For DataHub GMS resource look-up behavior please refer [Plugin Installation](#plugin-installation) section. Sample code of `getResourceAsStream` is available in sample Authenticator plugin [TestAuthenticator.java](../metadata-service/plugin/src/test/sample-test-plugins/src/main/java/com/datahub/plugins/test/TestAuthenticator.java).
 
-4. Bundle your Jar: Use `com.github.johnrengelman.shadow` gradle plugin to create an uber jar. 
+4. Bundle your Jar: Use `com.gradleup.shadow` gradle plugin to create an uber jar. 
 
    To see an example of building an uber jar, check out the `build.gradle` file for the apache-ranger-plugin file of [Apache Ranger Plugin](https://github.com/acryldata/datahub-ranger-auth-plugin/tree/main/apache-ranger-plugin) for reference.
 
    Exclude signature files as shown in below `shadowJar` task.
 
    ```groovy
-     apply plugin: 'com.github.johnrengelman.shadow';
+     apply plugin: 'com.gradleup.shadow';
      shadowJar {
          // Exclude com.datahub.plugins package and files related to jar signature   
          exclude "META-INF/*.RSA", "META-INF/*.SF","META-INF/*.DSA"

entity-registry/src/main/java/com/linkedin/metadata/aspect/plugins/validation/AspectValidationException.java

@@ -18,45 +18,39 @@ public static AspectValidationException forItem(BatchItem item, String msg) {
   }
 
   public static AspectValidationException forItem(BatchItem item, String msg, Exception e) {
-    return new AspectValidationException(
-        item.getChangeType(), item.getUrn(), item.getAspectName(), msg, SubType.VALIDATION, e);
+    return new AspectValidationException(item, msg, SubType.VALIDATION, e);
   }
 
   public static AspectValidationException forPrecondition(BatchItem item, String msg) {
     return forPrecondition(item, msg, null);
   }
 
+  public static AspectValidationException forFilter(BatchItem item, String msg) {
+    return new AspectValidationException(item, msg, SubType.FILTER);
+  }
+
   public static AspectValidationException forPrecondition(BatchItem item, String msg, Exception e) {
-    return new AspectValidationException(
-        item.getChangeType(), item.getUrn(), item.getAspectName(), msg, SubType.PRECONDITION, e);
+    return new AspectValidationException(item, msg, SubType.PRECONDITION, e);
   }
 
+  @Nonnull BatchItem item;
   @Nonnull ChangeType changeType;
   @Nonnull Urn entityUrn;
   @Nonnull String aspectName;
   @Nonnull SubType subType;
   @Nullable String msg;
 
-  public AspectValidationException(
-      @Nonnull ChangeType changeType,
-      @Nonnull Urn entityUrn,
-      @Nonnull String aspectName,
-      String msg,
-      SubType subType) {
-    this(changeType, entityUrn, aspectName, msg, subType, null);
+  public AspectValidationException(@Nonnull BatchItem item, String msg, SubType subType) {
+    this(item, msg, subType, null);
   }
 
   public AspectValidationException(
-      @Nonnull ChangeType changeType,
-      @Nonnull Urn entityUrn,
-      @Nonnull String aspectName,
-      @Nonnull String msg,
-      @Nullable SubType subType,
-      Exception e) {
+      @Nonnull BatchItem item, @Nonnull String msg, @Nullable SubType subType, Exception e) {
     super(msg, e);
-    this.changeType = changeType;
-    this.entityUrn = entityUrn;
-    this.aspectName = aspectName;
+    this.item = item;
+    this.changeType = item.getChangeType();
+    this.entityUrn = item.getUrn();
+    this.aspectName = item.getAspectName();
     this.msg = msg;
     this.subType = subType != null ? subType : SubType.VALIDATION;
   }
@@ -65,8 +59,12 @@ public Pair<Urn, String> getAspectGroup() {
     return Pair.of(entityUrn, aspectName);
   }
 
-  public static enum SubType {
+  public enum SubType {
+    // A validation exception is thrown
     VALIDATION,
-    PRECONDITION
+    // A failed precondition is thrown if the header constraints are not met
+    PRECONDITION,
+    // Exclude from processing further
+    FILTER
   }
 }

entity-registry/src/main/java/com/linkedin/metadata/aspect/plugins/validation/ValidationExceptionCollection.java

@@ -15,21 +15,38 @@
 public class ValidationExceptionCollection
     extends HashMap<Pair<Urn, String>, Set<AspectValidationException>> {
 
+  private final Set<Integer> failedHashCodes;
+  private final Set<Integer> filteredHashCodes;
+
+  public ValidationExceptionCollection() {
+    super();
+    this.failedHashCodes = new HashSet<>();
+    this.filteredHashCodes = new HashSet<>();
+  }
+
+  public boolean hasFatalExceptions() {
+    return !failedHashCodes.isEmpty();
+  }
+
   public static ValidationExceptionCollection newCollection() {
     return new ValidationExceptionCollection();
   }
 
   public void addException(AspectValidationException exception) {
     super.computeIfAbsent(exception.getAspectGroup(), key -> new HashSet<>()).add(exception);
+    if (!AspectValidationException.SubType.FILTER.equals(exception.getSubType())) {
+      failedHashCodes.add(exception.getItem().hashCode());
+    } else {
+      filteredHashCodes.add(exception.getItem().hashCode());
+    }
   }
 
   public void addException(BatchItem item, String message) {
     addException(item, message, null);
   }
 
   public void addException(BatchItem item, String message, Exception ex) {
-    super.computeIfAbsent(Pair.of(item.getUrn(), item.getAspectName()), key -> new HashSet<>())
-        .add(AspectValidationException.forItem(item, message, ex));
+    addException(AspectValidationException.forItem(item, message, ex));
   }
 
   public Stream<AspectValidationException> streamAllExceptions() {
@@ -41,15 +58,16 @@ public <T extends BatchItem> Collection<T> successful(Collection<T> items) {
   }
 
   public <T extends BatchItem> Stream<T> streamSuccessful(Stream<T> items) {
-    return items.filter(i -> !this.containsKey(Pair.of(i.getUrn(), i.getAspectName())));
+    return items.filter(
+        i -> !failedHashCodes.contains(i.hashCode()) && !filteredHashCodes.contains(i.hashCode()));
   }
 
   public <T extends BatchItem> Collection<T> exceptions(Collection<T> items) {
     return streamExceptions(items.stream()).collect(Collectors.toList());
   }
 
   public <T extends BatchItem> Stream<T> streamExceptions(Stream<T> items) {
-    return items.filter(i -> this.containsKey(Pair.of(i.getUrn(), i.getAspectName())));
+    return items.filter(i -> failedHashCodes.contains(i.hashCode()));
   }
 
   @Override

entity-registry/src/main/java/com/linkedin/metadata/aspect/validation/CreateIfNotExistsValidator.java

@@ -25,6 +25,8 @@
 @Getter
 @Accessors(chain = true)
 public class CreateIfNotExistsValidator extends AspectPayloadValidator {
+  public static final String FILTER_EXCEPTION_HEADER = "If-None-Match";
+  public static final String FILTER_EXCEPTION_VALUE = "*";
 
   @Nonnull private AspectPluginConfig config;
 
@@ -49,22 +51,34 @@ protected Stream<AspectValidationException> validatePreCommitAspects(
             .filter(item -> ChangeType.CREATE_ENTITY.equals(item.getChangeType()))
             .collect(Collectors.toSet())) {
       // if the key aspect is missing in the batch, the entity exists and CREATE_ENTITY should be
-      // denied
+      // denied or dropped
       if (!entityKeyMap.containsKey(createEntityItem.getUrn())) {
-        exceptions.addException(
-            createEntityItem,
-            "Cannot perform CREATE_ENTITY if not exists since the entity key already exists.");
+        if (isPrecondition(createEntityItem)) {
+          exceptions.addException(
+              AspectValidationException.forFilter(
+                  createEntityItem, "Dropping write per precondition header If-None-Match: *"));
+        } else {
+          exceptions.addException(
+              createEntityItem,
+              "Cannot perform CREATE_ENTITY if not exists since the entity key already exists.");
+        }
       }
     }
 
     for (ChangeMCP createItem :
         changeMCPs.stream()
             .filter(item -> ChangeType.CREATE.equals(item.getChangeType()))
             .collect(Collectors.toSet())) {
-      // if a CREATE item has a previous value, should be denied
+      // if a CREATE item has a previous value, should be denied or dropped
       if (createItem.getPreviousRecordTemplate() != null) {
-        exceptions.addException(
-            createItem, "Cannot perform CREATE since the aspect already exists.");
+        if (isPrecondition(createItem)) {
+          exceptions.addException(
+              AspectValidationException.forFilter(
+                  createItem, "Dropping write per precondition header If-None-Match: *"));
+        } else {
+          exceptions.addException(
+              createItem, "Cannot perform CREATE since the aspect already exists.");
+        }
       }
     }
 
@@ -77,4 +91,10 @@ protected Stream<AspectValidationException> validateProposedAspects(
       @Nonnull RetrieverContext retrieverContext) {
     return Stream.empty();
   }
+
+  private static boolean isPrecondition(ChangeMCP item) {
+    return item.getHeader(FILTER_EXCEPTION_HEADER)
+        .map(FILTER_EXCEPTION_VALUE::equals)
+        .orElse(false);
+  }
 }

entity-registry/src/testFixtures/java/com/linkedin/test/metadata/aspect/batch/TestMCP.java

@@ -21,6 +21,7 @@
 import com.linkedin.test.metadata.aspect.TestEntityRegistry;
 import java.net.URISyntaxException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -140,7 +141,7 @@ public Map<String, String> getHeaders() {
             mcp ->
                 mcp.getHeaders().entrySet().stream()
                     .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
-        .orElse(headers);
+        .orElse(headers != null ? headers : Collections.emptyMap());
   }
 
   @Override

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,8 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.0.2-bin.zip
+distributionSha256Sum=89d4e70e4e84e2d2dfbb63e4daa53e21b25017cc70c37e4eea31ee51fb15098a
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.11.1-all.zip
 networkTimeout=10000
+validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists