fix(system-update): sql setup regression (#15320)

Co-authored-by: David Leifker <david.leifker@acryl.io>

Author: chakru-r

.github/workflows/docker-unified-nightly.yml

@@ -323,6 +323,7 @@ jobs:
           TEST_STRATEGY: ${{ matrix.test_strategy }}
           BATCH_COUNT: "1" # since this workflow runs only on schedule trigger, batching isn't really needed.
           BATCH_NUMBER: "0"
+          PROFILE_NAME: ${{ matrix.profile }}
         run: |
           echo "$DATAHUB_VERSION"
           ./gradlew --stop

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/sqlsetup/CreateCdcUserStep.java

@@ -80,10 +80,12 @@ SqlSetupResult createCdcUser(SqlSetupArgs args) throws SQLException {
           stmt.executeUpdate();
         }
 
-        String grantCdcPrivilegesSql =
+        java.util.List<String> grantCdcPrivilegesStmts =
             dbOps.grantCdcPrivilegesSql(args.getCdcUser(), args.getDatabaseName());
-        try (PreparedStatement grantStmt = connection.prepareStatement(grantCdcPrivilegesSql)) {
-          grantStmt.executeUpdate();
+        for (String grantSql : grantCdcPrivilegesStmts) {
+          try (PreparedStatement grantStmt = connection.prepareStatement(grantSql)) {
+            grantStmt.executeUpdate();
+          }
         }
 
         result.setCdcUserCreated(true);

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/sqlsetup/DatabaseOperations.java

@@ -88,13 +88,14 @@ static DatabaseOperations create(DatabaseType dbType) {
   String createCdcUserSql(String cdcUser, String cdcPassword);
 
   /**
-   * Generate SQL for granting CDC privileges to a user.
+   * Generate SQL statements for granting CDC privileges to a user. Returns a list of individual SQL
+   * statements that can be executed separately.
    *
    * @param cdcUser the CDC username
    * @param databaseName the database name
-   * @return the SQL statement for granting CDC privileges
+   * @return list of SQL statements for granting CDC privileges
    */
-  String grantCdcPrivilegesSql(String cdcUser, String databaseName);
+  java.util.List<String> grantCdcPrivilegesSql(String cdcUser, String databaseName);
 
   /**
    * Generate SQL statements for creating the metadata_aspect_v2 table and its indexes. Returns a

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/sqlsetup/MySqlDatabaseOperations.java

@@ -78,17 +78,18 @@ public String createCdcUserSql(String cdcUser, String cdcPassword) {
   }
 
   @Override
-  public String grantCdcPrivilegesSql(String cdcUser, String databaseName) {
+  public java.util.List<String> grantCdcPrivilegesSql(String cdcUser, String databaseName) {
     // MySQL - comprehensive CDC privileges (matching original init-cdc.sql)
-    return String.format(
-        """
-        GRANT SELECT ON `%s`.* TO '%s'@'%%';
-        GRANT RELOAD ON *.* TO '%s'@'%%';
-        GRANT REPLICATION CLIENT ON *.* TO '%s'@'%%';
-        GRANT REPLICATION SLAVE ON *.* TO '%s'@'%%';
-        FLUSH PRIVILEGES;
-        """,
-        databaseName, cdcUser, cdcUser, cdcUser, cdcUser);
+    // Return as separate statements since JDBC doesn't support multiple statements in one execution
+    String escapedUser = escapeMysqlStringLiteral(cdcUser);
+    String escapedDatabase = escapeMysqlIdentifier(databaseName);
+
+    return java.util.Arrays.asList(
+        String.format("GRANT SELECT ON %s.* TO %s@'%%'", escapedDatabase, escapedUser),
+        String.format("GRANT RELOAD ON *.* TO %s@'%%'", escapedUser),
+        String.format("GRANT REPLICATION CLIENT ON *.* TO %s@'%%'", escapedUser),
+        String.format("GRANT REPLICATION SLAVE ON *.* TO %s@'%%'", escapedUser),
+        "FLUSH PRIVILEGES");
   }
 
   @Override

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/sqlsetup/PostgresDatabaseOperations.java

@@ -104,21 +104,23 @@ IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = %s) THEN
   }
 
   @Override
-  public String grantCdcPrivilegesSql(String cdcUser, String databaseName) {
+  public java.util.List<String> grantCdcPrivilegesSql(String cdcUser, String databaseName) {
     // PostgreSQL comprehensive CDC privileges (matching original init-cdc.sql)
-    return String.format(
-        """
-        GRANT CONNECT ON DATABASE "%s" TO "%s";
-        GRANT USAGE ON SCHEMA public TO "%s";
-        GRANT CREATE ON DATABASE "%s" TO "%s";
-        GRANT SELECT ON ALL TABLES IN SCHEMA public TO "%s";
-        ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO "%s";
-        ALTER USER "%s" WITH SUPERUSER;
-        ALTER TABLE public.metadata_aspect_v2 OWNER TO "%s";
-        ALTER TABLE public.metadata_aspect_v2 REPLICA IDENTITY FULL;
-        CREATE PUBLICATION dbz_publication FOR TABLE public.metadata_aspect_v2;
-        """,
-        databaseName, cdcUser, cdcUser, databaseName, cdcUser, cdcUser, cdcUser, cdcUser, cdcUser);
+    // Return as separate statements since JDBC doesn't support multiple statements in one execution
+    String escapedUser = escapePostgresIdentifier(cdcUser);
+    String escapedDatabase = escapePostgresIdentifier(databaseName);
+
+    return java.util.Arrays.asList(
+        String.format("GRANT CONNECT ON DATABASE %s TO %s", escapedDatabase, escapedUser),
+        String.format("GRANT USAGE ON SCHEMA public TO %s", escapedUser),
+        String.format("GRANT CREATE ON DATABASE %s TO %s", escapedDatabase, escapedUser),
+        String.format("GRANT SELECT ON ALL TABLES IN SCHEMA public TO %s", escapedUser),
+        String.format(
+            "ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO %s", escapedUser),
+        String.format("ALTER USER %s WITH SUPERUSER", escapedUser),
+        String.format("ALTER TABLE public.metadata_aspect_v2 OWNER TO %s", escapedUser),
+        "ALTER TABLE public.metadata_aspect_v2 REPLICA IDENTITY FULL",
+        "CREATE PUBLICATION dbz_publication FOR TABLE public.metadata_aspect_v2");
   }
 
   @Override

datahub-upgrade/src/test/java/com/linkedin/datahub/upgrade/sqlsetup/CreateCdcUserStepTest.java

@@ -478,9 +478,10 @@ public void testCreateCdcUserSuccess() throws SQLException {
     assertEquals(result.isCdcUserCreated(), true);
     assertTrue(result.getExecutionTimeMs() >= 0);
     // Verify PreparedStatement approach - should call dataSource() and prepareStatement()
+    // MySQL: 1 createCdcUser + 5 grantCdcPrivileges statements = 6 total
     verify(mockDatabase).dataSource();
-    verify(mockConnection, times(2)).prepareStatement(anyString());
-    verify(mockPreparedStatement, times(2)).executeUpdate();
+    verify(mockConnection, times(6)).prepareStatement(anyString());
+    verify(mockPreparedStatement, times(6)).executeUpdate();
   }
 
   @Test
@@ -533,9 +534,10 @@ public void testCreateCdcUserWithPostgresComplexSql() throws SQLException {
     assertEquals(result.isCdcUserCreated(), true);
     assertTrue(result.getExecutionTimeMs() >= 0);
     // Verify PreparedStatement approach - should call dataSource() and prepareStatement()
+    // MySQL: 1 createCdcUser + 5 grantCdcPrivileges statements = 6 total
     verify(mockDatabase).dataSource();
-    verify(mockConnection, times(2)).prepareStatement(anyString());
-    verify(mockPreparedStatement, times(2)).executeUpdate();
+    verify(mockConnection, times(6)).prepareStatement(anyString());
+    verify(mockPreparedStatement, times(6)).executeUpdate();
   }
 
   @Test
@@ -562,8 +564,9 @@ public void testCreateCdcUserWithMysqlSimpleSql() throws SQLException {
     assertEquals(result.isCdcUserCreated(), true);
     assertTrue(result.getExecutionTimeMs() >= 0);
     // Verify PreparedStatement approach - should call dataSource() and prepareStatement()
+    // MySQL: 1 createCdcUser + 5 grantCdcPrivileges statements = 6 total
     verify(mockDatabase).dataSource();
-    verify(mockConnection, times(2)).prepareStatement(anyString());
-    verify(mockPreparedStatement, times(2)).executeUpdate();
+    verify(mockConnection, times(6)).prepareStatement(anyString());
+    verify(mockPreparedStatement, times(6)).executeUpdate();
   }
 }

datahub-upgrade/src/test/java/com/linkedin/datahub/upgrade/sqlsetup/DatabaseOperationsTest.java

@@ -113,16 +113,22 @@ public void testMysqlCreateCdcUserSql() {
 
   @Test
   public void testPostgresGrantCdcPrivilegesSql() {
-    String result = postgresOps.grantCdcPrivilegesSql("cdcuser", "testdb");
-    assertTrue(result.contains("GRANT CONNECT ON DATABASE \"testdb\" TO \"cdcuser\""));
-    assertTrue(result.contains("CREATE PUBLICATION dbz_publication"));
+    java.util.List<String> statements = postgresOps.grantCdcPrivilegesSql("cdcuser", "testdb");
+    assertNotNull(statements);
+    assertTrue(statements.size() > 0);
+    String allStatements = String.join(" ", statements);
+    assertTrue(allStatements.contains("GRANT CONNECT ON DATABASE \"testdb\" TO \"cdcuser\""));
+    assertTrue(allStatements.contains("CREATE PUBLICATION dbz_publication"));
   }
 
   @Test
   public void testMysqlGrantCdcPrivilegesSql() {
-    String result = mysqlOps.grantCdcPrivilegesSql("cdcuser", "testdb");
-    assertTrue(result.contains("GRANT SELECT ON `testdb`.* TO 'cdcuser'@'%'"));
-    assertTrue(result.contains("GRANT REPLICATION CLIENT ON *.* TO 'cdcuser'@'%'"));
+    java.util.List<String> statements = mysqlOps.grantCdcPrivilegesSql("cdcuser", "testdb");
+    assertNotNull(statements);
+    assertTrue(statements.size() > 0);
+    String allStatements = String.join(" ", statements);
+    assertTrue(allStatements.contains("GRANT SELECT ON `testdb`.* TO 'cdcuser'@'%'"));
+    assertTrue(allStatements.contains("GRANT REPLICATION CLIENT ON *.* TO 'cdcuser'@'%'"));
   }
 
   @Test

docker/profiles/docker-compose.gms.yml

@@ -76,6 +76,11 @@ x-datahub-system-update-service: &datahub-system-update-service
     - ${DATAHUB_LOCAL_SYS_UPDATE_ENV:-empty2.env}
   environment: &datahub-system-update-env
     <<: [*primary-datastore-mysql-env, *graph-datastore-search-env, *search-datastore-env, *kafka-env, *datahub-common-env]
+    EBEAN_DATASOURCE_USERNAME: ${DATAHUB_EBEAN_DATASOURCE_ROOT_USERNAME:-root}
+    EBEAN_DATASOURCE_PASSWORD: ${DATAHUB_EBEAN_DATASOURCE_ROOT_PASSWORD:-datahub}
+    CREATE_USER: ${DATAHUB_CREATE_USER:-false}
+    CREATE_USER_USERNAME: ${DATAHUB_CREATE_USER_USERNAME:-datahub}
+    CREATE_USER_PASSWORD: ${DATAHUB_CREATE_USER_PASSWORD:-datahub}
     SCHEMA_REGISTRY_SYSTEM_UPDATE: ${SCHEMA_REGISTRY_SYSTEM_UPDATE:-true}
     SPRING_KAFKA_PROPERTIES_AUTO_REGISTER_SCHEMAS: ${SPRING_KAFKA_PROPERTIES_AUTO_REGISTER_SCHEMAS:-true}
     SPRING_KAFKA_PROPERTIES_USE_LATEST_VERSION: ${SPRING_KAFKA_PROPERTIES_USE_LATEST_VERSION:-true}

docker/quickstart/docker-compose.quickstart-profile.yml

@@ -406,6 +406,9 @@ services:
         required: true
     environment:
       BACKFILL_BROWSE_PATHS_V2: 'true'
+      CREATE_USER: 'false'
+      CREATE_USER_PASSWORD: datahub
+      CREATE_USER_USERNAME: datahub
       DATAHUB_BASE_PATH: /
       DATAHUB_GMS_BASE_PATH: /
       DATAHUB_GMS_HOST: datahub-gms
@@ -416,7 +419,7 @@ services:
       EBEAN_DATASOURCE_HOST: mysql:3306
       EBEAN_DATASOURCE_PASSWORD: datahub
       EBEAN_DATASOURCE_URL: jdbc:mysql://mysql:3306/datahub?verifyServerCertificate=false&useSSL=true&useUnicode=yes&characterEncoding=UTF-8&enabledTLSProtocols=TLSv1.2
-      EBEAN_DATASOURCE_USERNAME: datahub
+      EBEAN_DATASOURCE_USERNAME: root
       ELASTICSEARCH_BUILD_INDICES_CLONE_INDICES: 'false'
       ELASTICSEARCH_HOST: search
       ELASTICSEARCH_IMPLEMENTATION: opensearch

docs/how/updating-datahub.md

@@ -8,6 +8,7 @@
 
 - #13726: Default search results per page (new: 5000, old: 10000) can be configured with environment variable `ELASTICSEARCH_LIMIT_RESULTS_API_DEFAULT`
 - #13726: Maximum lineage visualization hops (new: 20, old: 1000) can be configured with environment variable `ELASTICSEARCH_SEARCH_GRAPH_LINEAGE_MAX_HOPS`
+- #15044: If using the new system-update sqlSetup, the system-update job must be granted privileges to create users, databases, and tables.
 
 ### Known Issues