Merge branch 'stacktrace-in-looker-log' of github.com:sid-acryl/datah…

…ub-fork into stacktrace-in-looker-log

.github/scripts/check_python_package.py

@@ -1,18 +1,33 @@
 import setuptools
+import os
 
 folders = ["./smoke-test/tests"]
 
 for folder in folders:
     print(f"Checking folder {folder}")
-    a = [i for i in setuptools.find_packages(folder) if "cypress" not in i]
-    b = [i for i in setuptools.find_namespace_packages(folder) if "cypress" not in i]
+    packages = [i for i in setuptools.find_packages(folder) if "cypress" not in i]
+    namespace_packages = [
+        i for i in setuptools.find_namespace_packages(folder) if "cypress" not in i
+    ]
 
-    in_a_not_b = set(a) - set(b)
-    in_b_not_a = set(b) - set(a)
+    print("Packages found:", packages)
+    print("Namespace packages found:", namespace_packages)
+
+    in_packages_not_namespace = set(packages) - set(namespace_packages)
+    in_namespace_not_packages = set(namespace_packages) - set(packages)
+
+    if in_packages_not_namespace:
+        print(f"Packages not in namespace packages: {in_packages_not_namespace}")
+    if in_namespace_not_packages:
+        print(f"Namespace packages not in packages: {in_namespace_not_packages}")
+        for pkg in in_namespace_not_packages:
+            pkg_path = os.path.join(folder, pkg.replace(".", os.path.sep))
+            print(f"Contents of {pkg_path}:")
+            print(os.listdir(pkg_path))
 
     assert (
-        len(in_a_not_b) == 0
-    ), f"Found packages in {folder} that are not in namespace packages: {in_a_not_b}"
+        len(in_packages_not_namespace) == 0
+    ), f"Found packages in {folder} that are not in namespace packages: {in_packages_not_namespace}"
     assert (
-        len(in_b_not_a) == 0
-    ), f"Found namespace packages in {folder} that are not in packages: {in_b_not_a}"
+        len(in_namespace_not_packages) == 0
+    ), f"Found namespace packages in {folder} that are not in packages: {in_namespace_not_packages}"

build.gradle

@@ -54,7 +54,7 @@ buildscript {
   ext.hazelcastVersion = '5.3.6'
   ext.ebeanVersion = '12.16.1'
   ext.googleJavaFormatVersion = '1.18.1'
-  ext.openLineageVersion = '1.16.0'
+  ext.openLineageVersion = '1.19.0'
   ext.logbackClassicJava8 = '1.2.12'
 
   ext.docker_registry = 'acryldata'
@@ -111,6 +111,7 @@ project.ext.externalDependency = [
     'avroCompiler': 'org.apache.avro:avro-compiler:1.11.3',
     'awsGlueSchemaRegistrySerde': 'software.amazon.glue:schema-registry-serde:1.1.17',
     'awsMskIamAuth': 'software.amazon.msk:aws-msk-iam-auth:2.0.3',
+    'awsS3': 'software.amazon.awssdk:s3:2.26.21',
     'awsSecretsManagerJdbc': 'com.amazonaws.secretsmanager:aws-secretsmanager-jdbc:1.0.13',
     'awsPostgresIamAuth': 'software.amazon.jdbc:aws-advanced-jdbc-wrapper:1.0.2',
     'awsRds':'software.amazon.awssdk:rds:2.18.24',

datahub-graphql-core/build.gradle

@@ -22,6 +22,7 @@ dependencies {
     implementation externalDependency.opentelemetryAnnotations
 
     implementation externalDependency.slf4jApi
+    implementation externalDependency.springContext
     compileOnly externalDependency.lombok
     annotationProcessor externalDependency.lombok
 

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

@@ -24,6 +24,7 @@ private Constants() {}
   public static final String PROPERTIES_SCHEMA_FILE = "properties.graphql";
   public static final String FORMS_SCHEMA_FILE = "forms.graphql";
   public static final String ASSERTIONS_SCHEMA_FILE = "assertions.graphql";
+  public static final String COMMON_SCHEMA_FILE = "common.graphql";
   public static final String INCIDENTS_SCHEMA_FILE = "incident.graphql";
   public static final String CONTRACTS_SCHEMA_FILE = "contract.graphql";
   public static final String CONNECTIONS_SCHEMA_FILE = "connection.graphql";

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

@@ -96,6 +96,7 @@
 import com.linkedin.datahub.graphql.generated.MLPrimaryKey;
 import com.linkedin.datahub.graphql.generated.MLPrimaryKeyProperties;
 import com.linkedin.datahub.graphql.generated.MatchedField;
+import com.linkedin.datahub.graphql.generated.MetadataAttribution;
 import com.linkedin.datahub.graphql.generated.Notebook;
 import com.linkedin.datahub.graphql.generated.Owner;
 import com.linkedin.datahub.graphql.generated.OwnershipTypeEntity;
@@ -284,6 +285,8 @@
 import com.linkedin.datahub.graphql.resolvers.search.SearchAcrossEntitiesResolver;
 import com.linkedin.datahub.graphql.resolvers.search.SearchAcrossLineageResolver;
 import com.linkedin.datahub.graphql.resolvers.search.SearchResolver;
+import com.linkedin.datahub.graphql.resolvers.settings.docPropagation.DocPropagationSettingsResolver;
+import com.linkedin.datahub.graphql.resolvers.settings.docPropagation.UpdateDocPropagationSettingsResolver;
 import com.linkedin.datahub.graphql.resolvers.settings.user.UpdateCorpUserViewsSettingsResolver;
 import com.linkedin.datahub.graphql.resolvers.settings.view.GlobalViewsSettingsResolver;
 import com.linkedin.datahub.graphql.resolvers.settings.view.UpdateGlobalViewsSettingsResolver;
@@ -695,7 +698,8 @@ public GmsGraphQLEngine(final GmsGraphQLEngineArgs args) {
                 businessAttributeType));
     this.loadableTypes = new ArrayList<>(entityTypes);
     // Extend loadable types with types from the plugins
-    // This allows us to offer search and browse capabilities out of the box for those types
+    // This allows us to offer search and browse capabilities out of the box for
+    // those types
     for (GmsGraphQLPlugin plugin : this.graphQLPlugins) {
       this.entityTypes.addAll(plugin.getEntityTypes());
       Collection<? extends LoadableType<?, ?>> pluginLoadableTypes = plugin.getLoadableTypes();
@@ -790,6 +794,7 @@ public void configureRuntimeWiring(final RuntimeWiring.Builder builder) {
     configureBusinessAttributeAssociationResolver(builder);
     configureConnectionResolvers(builder);
     configureDeprecationResolvers(builder);
+    configureMetadataAttributionResolver(builder);
   }
 
   private void configureOrganisationRoleResolvers(RuntimeWiring.Builder builder) {
@@ -843,7 +848,8 @@ public GraphQLEngine.Builder builder() {
         .addSchema(fileBasedSchema(CONNECTIONS_SCHEMA_FILE))
         .addSchema(fileBasedSchema(ASSERTIONS_SCHEMA_FILE))
         .addSchema(fileBasedSchema(INCIDENTS_SCHEMA_FILE))
-        .addSchema(fileBasedSchema(CONTRACTS_SCHEMA_FILE));
+        .addSchema(fileBasedSchema(CONTRACTS_SCHEMA_FILE))
+        .addSchema(fileBasedSchema(COMMON_SCHEMA_FILE));
 
     for (GmsGraphQLPlugin plugin : this.graphQLPlugins) {
       List<String> pluginSchemaFiles = plugin.getSchemaFiles();
@@ -1087,8 +1093,10 @@ private void configureQueryResolvers(final RuntimeWiring.Builder builder) {
                     new BrowseV2Resolver(this.entityClient, this.viewService, this.formService))
                 .dataFetcher("businessAttribute", getResolver(businessAttributeType))
                 .dataFetcher(
-                    "listBusinessAttributes",
-                    new ListBusinessAttributesResolver(this.entityClient)));
+                    "listBusinessAttributes", new ListBusinessAttributesResolver(this.entityClient))
+                .dataFetcher(
+                    "docPropagationSettings",
+                    new DocPropagationSettingsResolver(this.settingsService)));
   }
 
   private DataFetcher getEntitiesResolver() {
@@ -1340,7 +1348,11 @@ private void configureMutationResolvers(final RuntimeWiring.Builder builder) {
               .dataFetcher(
                   "createForm", new CreateFormResolver(this.entityClient, this.formService))
               .dataFetcher("deleteForm", new DeleteFormResolver(this.entityClient))
-              .dataFetcher("updateForm", new UpdateFormResolver(this.entityClient));
+              .dataFetcher("updateForm", new UpdateFormResolver(this.entityClient))
+              .dataFetcher(
+                  "updateDocPropagationSettings",
+                  new UpdateDocPropagationSettingsResolver(this.settingsService));
+
           if (featureFlags.isBusinessAttributeEntityEnabled()) {
             typeWiring
                 .dataFetcher(
@@ -1869,7 +1881,9 @@ private void configureCorpGroupResolvers(final RuntimeWiring.Builder builder) {
         "CorpGroup",
         typeWiring ->
             typeWiring
-                .dataFetcher("relationships", new EntityRelationshipsResultResolver(graphClient))
+                .dataFetcher(
+                    "relationships",
+                    new EntityRelationshipsResultResolver(graphClient, entityService))
                 .dataFetcher("privileges", new EntityPrivilegesResolver(entityClient))
                 .dataFetcher(
                     "aspects", new WeaklyTypedAspectsResolver(entityClient, entityRegistry))
@@ -2821,7 +2835,8 @@ private void configureContractResolvers(final RuntimeWiring.Builder builder) {
   }
 
   private void configurePolicyResolvers(final RuntimeWiring.Builder builder) {
-    // Register resolvers for "resolvedUsers" and "resolvedGroups" field of the Policy type.
+    // Register resolvers for "resolvedUsers" and "resolvedGroups" field of the
+    // Policy type.
     builder.type(
         "ActorFilter",
         typeWiring ->
@@ -3174,4 +3189,20 @@ private void configureDeprecationResolvers(final RuntimeWiring.Builder builder)
                 new EntityTypeResolver(
                     entityTypes, (env) -> ((Deprecation) env.getSource()).getActorEntity())));
   }
+
+  private void configureMetadataAttributionResolver(final RuntimeWiring.Builder builder) {
+    builder.type(
+        "MetadataAttribution",
+        typeWiring ->
+            typeWiring
+                .dataFetcher(
+                    "actor",
+                    new EntityTypeResolver(
+                        entityTypes, (env) -> ((MetadataAttribution) env.getSource()).getActor()))
+                .dataFetcher(
+                    "source",
+                    new EntityTypeResolver(
+                        entityTypes,
+                        (env) -> ((MetadataAttribution) env.getSource()).getSource())));
+  }
 }

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/authorization/AuthorizationUtils.java

@@ -398,6 +398,11 @@ public static boolean canManageForms(@Nonnull QueryContext context) {
         PoliciesConfig.MANAGE_DOCUMENTATION_FORMS_PRIVILEGE);
   }
 
+  public static boolean canManageFeatures(@Nonnull QueryContext context) {
+    return AuthUtil.isAuthorized(
+        context.getAuthorizer(), context.getActorUrn(), PoliciesConfig.MANAGE_FEATURES_PRIVILEGE);
+  }
+
   public static boolean isAuthorized(
       @Nonnull Authorizer authorizer,
       @Nonnull String actor,

datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/ingest/source/UpsertIngestionSourceResolver.java

@@ -25,12 +25,16 @@
 import graphql.schema.DataFetcher;
 import graphql.schema.DataFetchingEnvironment;
 import java.net.URISyntaxException;
+import java.time.DateTimeException;
+import java.time.ZoneId;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.support.CronExpression;
 
 /** Creates or updates an ingestion source. Requires the MANAGE_INGESTION privilege. */
 @Slf4j
@@ -46,55 +50,51 @@ public UpsertIngestionSourceResolver(final EntityClient entityClient) {
   public CompletableFuture<String> get(final DataFetchingEnvironment environment) throws Exception {
     final QueryContext context = environment.getContext();
 
-    return GraphQLConcurrencyUtils.supplyAsync(
-        () -> {
-          if (IngestionAuthUtils.canManageIngestion(context)) {
-
-            final Optional<String> ingestionSourceUrn =
-                Optional.ofNullable(environment.getArgument("urn"));
-            final UpdateIngestionSourceInput input =
-                bindArgument(environment.getArgument("input"), UpdateIngestionSourceInput.class);
+    if (!IngestionAuthUtils.canManageIngestion(context)) {
+      throw new AuthorizationException(
+          "Unauthorized to perform this action. Please contact your DataHub administrator.");
+    }
+    final Optional<String> ingestionSourceUrn = Optional.ofNullable(environment.getArgument("urn"));
+    final UpdateIngestionSourceInput input =
+        bindArgument(environment.getArgument("input"), UpdateIngestionSourceInput.class);
 
-            // Create the policy info.
-            final DataHubIngestionSourceInfo info = mapIngestionSourceInfo(input);
-            final MetadataChangeProposal proposal;
-            if (ingestionSourceUrn.isPresent()) {
-              // Update existing ingestion source
-              try {
-                proposal =
-                    buildMetadataChangeProposalWithUrn(
-                        Urn.createFromString(ingestionSourceUrn.get()),
-                        INGESTION_INFO_ASPECT_NAME,
-                        info);
-              } catch (URISyntaxException e) {
-                throw new DataHubGraphQLException(
-                    String.format("Malformed urn %s provided.", ingestionSourceUrn.get()),
-                    DataHubGraphQLErrorCode.BAD_REQUEST);
-              }
-            } else {
-              // Create new ingestion source
-              // Since we are creating a new Ingestion Source, we need to generate a unique UUID.
-              final UUID uuid = UUID.randomUUID();
-              final String uuidStr = uuid.toString();
-              final DataHubIngestionSourceKey key = new DataHubIngestionSourceKey();
-              key.setId(uuidStr);
-              proposal =
-                  buildMetadataChangeProposalWithKey(
-                      key, INGESTION_SOURCE_ENTITY_NAME, INGESTION_INFO_ASPECT_NAME, info);
-            }
+    // Create the policy info.
+    final DataHubIngestionSourceInfo info = mapIngestionSourceInfo(input);
+    final MetadataChangeProposal proposal;
+    if (ingestionSourceUrn.isPresent()) {
+      // Update existing ingestion source
+      try {
+        proposal =
+            buildMetadataChangeProposalWithUrn(
+                Urn.createFromString(ingestionSourceUrn.get()), INGESTION_INFO_ASPECT_NAME, info);
+      } catch (URISyntaxException e) {
+        throw new DataHubGraphQLException(
+            String.format("Malformed urn %s provided.", ingestionSourceUrn.get()),
+            DataHubGraphQLErrorCode.BAD_REQUEST);
+      }
+    } else {
+      // Create new ingestion source
+      // Since we are creating a new Ingestion Source, we need to generate a unique UUID.
+      final UUID uuid = UUID.randomUUID();
+      final String uuidStr = uuid.toString();
+      final DataHubIngestionSourceKey key = new DataHubIngestionSourceKey();
+      key.setId(uuidStr);
+      proposal =
+          buildMetadataChangeProposalWithKey(
+              key, INGESTION_SOURCE_ENTITY_NAME, INGESTION_INFO_ASPECT_NAME, info);
+    }
 
-            try {
-              return _entityClient.ingestProposal(context.getOperationContext(), proposal, false);
-            } catch (Exception e) {
-              throw new RuntimeException(
-                  String.format(
-                      "Failed to perform update against ingestion source with urn %s",
-                      input.toString()),
-                  e);
-            }
+    return GraphQLConcurrencyUtils.supplyAsync(
+        () -> {
+          try {
+            return _entityClient.ingestProposal(context.getOperationContext(), proposal, false);
+          } catch (Exception e) {
+            throw new RuntimeException(
+                String.format(
+                    "Failed to perform update against ingestion source with urn %s",
+                    input.toString()),
+                e);
           }
-          throw new AuthorizationException(
-              "Unauthorized to perform this action. Please contact your DataHub administrator.");
         },
         this.getClass().getSimpleName(),
         "get");
@@ -137,9 +137,38 @@ private DataHubIngestionSourceConfig mapConfig(final UpdateIngestionSourceConfig
 
   private DataHubIngestionSourceSchedule mapSchedule(
       final UpdateIngestionSourceScheduleInput input) {
+
+    final String modifiedCronInterval = adjustCronInterval(input.getInterval());
+    try {
+      CronExpression.parse(modifiedCronInterval);
+    } catch (IllegalArgumentException e) {
+      throw new DataHubGraphQLException(
+          String.format("Invalid cron schedule `%s`: %s", input.getInterval(), e.getMessage()),
+          DataHubGraphQLErrorCode.BAD_REQUEST);
+    }
+    try {
+      ZoneId.of(input.getTimezone());
+    } catch (DateTimeException e) {
+      throw new DataHubGraphQLException(
+          String.format("Invalid timezone `%s`: %s", input.getTimezone(), e.getMessage()),
+          DataHubGraphQLErrorCode.BAD_REQUEST);
+    }
+
     final DataHubIngestionSourceSchedule result = new DataHubIngestionSourceSchedule();
     result.setInterval(input.getInterval());
     result.setTimezone(input.getTimezone());
     return result;
   }
+
+  // Copied from IngestionScheduler.java
+  private String adjustCronInterval(final String origCronInterval) {
+    Objects.requireNonNull(origCronInterval, "origCronInterval must not be null");
+    // Typically we support 5-character cron. Spring's lib only supports 6 character cron so we make
+    // an adjustment here.
+    final String[] originalCronParts = origCronInterval.split(" ");
+    if (originalCronParts.length == 5) {
+      return String.format("0 %s", origCronInterval);
+    }
+    return origCronInterval;
+  }
 }