branch-4.2: [fix](fe) Require ADMIN for remote CDC driver URLs #69617 (#70086)

Cherry-picked from #69617

fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java

@@ -77,6 +77,7 @@ import org.apache.doris.qe.ShowResultSetMetaData;
 import org.apache.doris.rpc.RpcException;
 import org.apache.doris.service.FrontendOptions;
 import org.apache.doris.system.Backend;
+import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
 import org.apache.doris.tablefunction.S3TableValuedFunction;
 import org.apache.doris.thrift.TCell;
 import org.apache.doris.thrift.TRow;
@@ -264,6 +265,8 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
         try {
             init();
             checkRequiredSourceProperties();
+            StreamingJobUtils.checkCdcDriverPrivilege(ConnectContext.get(),
+                    sourceProperties.get(DataSourceConfigKeys.DRIVER_URL));
             DataSourceConfigValidator.validateSourceBeforeTableCreation(
                     dataSourceType, sourceProperties);
             List<String> createTbls = createTableIfNotExists();
@@ -421,9 +424,13 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
     private void initInsertJob() {
         try {
             init();
+            ConnectContext ctx = ConnectContext.get();
             UnboundTVFRelation currentTvf = getCurrentTvf();
             this.tvfType = currentTvf.getFunctionName();
             this.originTvfProps = currentTvf.getProperties().getMap();
+            if (CdcStreamTableValuedFunction.NAME.equalsIgnoreCase(tvfType)) {
+                StreamingJobUtils.checkCdcDriverPrivilege(ctx, originTvfProps.get(DataSourceConfigKeys.DRIVER_URL));
+            }
             this.offsetProvider = createOffsetProvider(sourceProperties);
             this.offsetProvider.ensureInitialized(getJobId(), originTvfProps);
             // cdc_stream TVF has TABLE; S3 TVF doesn't, so syncTables stays null there.
@@ -481,9 +488,18 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
     public void initLogicalPlan(boolean regen) {
         if (regen || baseCommand == null) {
             this.baseCommand = null;
-            makeConnectContext();
-            LogicalPlan logicalPlan = new NereidsParser().parseSingle(getExecuteSql());
-            this.baseCommand = (InsertIntoTableCommand) logicalPlan;
+            ConnectContext previousContext = ConnectContext.get();
+            try {
+                makeConnectContext();
+                LogicalPlan logicalPlan = new NereidsParser().parseSingle(getExecuteSql());
+                this.baseCommand = (InsertIntoTableCommand) logicalPlan;
+            } finally {
+                // Subsequent privilege checks must retain the caller's identity and session roles.
+                ConnectContext.remove();
+                if (previousContext != null) {
+                    previousContext.setThreadLocalInfo();
+                }
+            }
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java

@@ -30,6 +30,7 @@ import org.apache.doris.catalog.Type;
 import org.apache.doris.cloud.system.CloudSystemInfoService;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.util.SmallFileMgr;
 import org.apache.doris.common.util.SmallFileMgr.SmallFile;
@@ -43,6 +44,8 @@ import org.apache.doris.job.common.DataSourceType;
 import org.apache.doris.job.exception.JobException;
 import org.apache.doris.job.extensions.insert.streaming.PostgresResourceValidator;
 import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
@@ -265,6 +268,15 @@ public class StreamingJobUtils {
         return ctx;
     }
 
+    public static void checkCdcDriverPrivilege(ConnectContext ctx, String driverUrl) {
+        if ((StringUtils.startsWithIgnoreCase(driverUrl, "http://")
+                || StringUtils.startsWithIgnoreCase(driverUrl, "https://"))
+                && !Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+            throw new AnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR.formatErrorMsg("ADMIN")
+                    + " HTTP/HTTPS CDC driver URLs require ADMIN privilege. Use a preinstalled JDBC driver instead.");
+        }
+    }
+
     public static JdbcClient getJdbcClient(DataSourceType sourceType, Map<String, String> properties) {
         JdbcClientConfig config = new JdbcClientConfig();
         config.setCatalog(sourceType.name());

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java

@@ -28,6 +28,7 @@ import org.apache.doris.job.common.JobStatus;
 import org.apache.doris.job.extensions.insert.streaming.DataSourceConfigValidator;
 import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
 import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
+import org.apache.doris.job.util.StreamingJobUtils;
 import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
 import org.apache.doris.nereids.parser.NereidsParser;
 import org.apache.doris.nereids.trees.plans.PlanType;
@@ -168,6 +169,8 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne
                 boolean sourcePropModified =
                         isPropertiesModified(streamingJob.getSourceProperties(), this.getSourceProperties());
                 if (sourcePropModified) {
+                    StreamingJobUtils.checkCdcDriverPrivilege(ConnectContext.get(),
+                            sourceProperties.get(DataSourceConfigKeys.DRIVER_URL));
                     DataSourceConfigValidator.validateSource(this.getSourceProperties(),
                             streamingJob.getDataSourceType().name());
                     checkUnmodifiableSourceProperties(streamingJob.getSourceProperties());
@@ -325,7 +328,8 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne
                                     inputTvf.getProperties().getMap().get(unmodifiable)),
                             "The '%s' property cannot be modified in ALTER JOB for cdc_stream", unmodifiable);
                 }
-                CdcStreamTableValuedFunction.validateProperties(inputTvf.getProperties().getMap());
+                // Validate merged resource properties as the ALTER caller, before schema discovery.
+                new CdcStreamTableValuedFunction(inputTvf.getProperties().getMap()).checkAuth(ConnectContext.get());
                 break;
             default:
                 throw new IllegalArgumentException("Unsupported tvf type:" + inputTvf.getFunctionName());

fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java

@@ -29,6 +29,7 @@ import org.apache.doris.job.common.DataSourceType;
 import org.apache.doris.job.extensions.insert.streaming.DataSourceConfigValidator;
 import org.apache.doris.job.extensions.insert.streaming.StreamingJdbcUrlNormalizer;
 import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.thrift.TBrokerFileStatus;
 import org.apache.doris.thrift.TFileType;
 
@@ -83,6 +84,8 @@ public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunctio
         StreamingJobUtils.populateDefaultSourceProperties(sourceType, copyProps, jobId);
 
         super.parseCommonProperties(copyProps);
+        StreamingJobUtils.checkCdcDriverPrivilege(ConnectContext.get(),
+                processedParams.get(DataSourceConfigKeys.DRIVER_URL));
         this.processedParams.put(ENABLE_CDC_CLIENT_KEY, "true");
         this.processedParams.put(URI_KEY, URI);
         this.processedParams.put(HTTP_ENABLE_RANGE_REQUEST_KEY, "false");

fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java

@@ -17,10 +17,13 @@
 
 package org.apache.doris.job.extensions.insert.streaming;
 
+import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
 import org.apache.doris.job.cdc.request.CommitOffsetRequest;
 import org.apache.doris.job.cdc.split.SnapshotSplit;
@@ -34,8 +37,10 @@ import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
 import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
 import org.apache.doris.job.offset.s3.S3Offset;
 import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
+import org.apache.doris.nereids.exceptions.ParseException;
 import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
 import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
 import org.apache.doris.tablefunction.S3TableValuedFunction;
 import org.apache.doris.transaction.GlobalTransactionMgrIface;
@@ -122,21 +127,49 @@ public class StreamingInsertJobOffsetPersistenceTest {
         Deencapsulation.setField(job, "jobId", 9001L);
         job.setJobStatus(JobStatus.PAUSED);
         Deencapsulation.setField(job, "tvfType", CdcStreamTableValuedFunction.NAME);
-        HashMap<String, String> tvfProperties = new HashMap<>();
-        tvfProperties.put(DataSourceConfigKeys.TYPE, "postgres");
-        tvfProperties.put(DataSourceConfigKeys.JDBC_URL, "jdbc:postgresql://127.0.0.1:5432/source_db");
-        tvfProperties.put(DataSourceConfigKeys.TABLE, "source_table");
-        Deencapsulation.setField(job, "originTvfProps", tvfProperties);
+        Deencapsulation.setField(job, "createUser", UserIdentity.ROOT);
+        job.setExecuteSql("INSERT INTO internal.test_db.target_table SELECT * FROM cdc_stream("
+                + "\"type\"=\"postgres\", \"jdbc_url\"=\"jdbc:postgresql://127.0.0.1:5432/source_db\", "
+                + "\"table\"=\"source_table\")");
 
         job.gsonPostProcess();
         JdbcTvfSourceOffsetProvider provider = Deencapsulation.getField(job, "offsetProvider");
         Assert.assertNull(Deencapsulation.getField(provider, "sourceType"));
 
-        job.validateAlterOffset("{\"lsn\":\"300\"}");
+        ConnectContext previousContext = ConnectContext.get();
+        try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class)) {
+            Env env = Mockito.mock(Env.class);
+            InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+            CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+            Mockito.when(catalog.getName()).thenReturn(InternalCatalog.INTERNAL_CATALOG_NAME);
+            Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+            Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+            Mockito.when(catalogMgr.getCatalog(InternalCatalog.INTERNAL_CATALOG_NAME)).thenReturn(catalog);
+            envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+            ConnectContext caller = new ConnectContext();
+            caller.setCurrentUserIdentity(UserIdentity.createAnalyzedUserIdentWithIp("alter_user", "%"));
+            caller.setAuthenticatedRoles(Collections.singleton("session_role"));
+            caller.setThreadLocalInfo();
 
-        Assert.assertEquals(JobStatus.PAUSED, job.getJobStatus());
-        Assert.assertThrows(AnalysisException.class,
-                () -> job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}"));
+            // Restored jobs have no cached plan/TVF properties. Parsing must not replace the caller.
+            job.validateAlterOffset("{\"lsn\":\"300\"}");
+
+            Assert.assertSame(caller, ConnectContext.get());
+            Assert.assertEquals(Collections.singleton("session_role"),
+                    ConnectContext.get().getAuthenticatedRoles());
+            Assert.assertEquals(JobStatus.PAUSED, job.getJobStatus());
+            Assert.assertThrows(AnalysisException.class,
+                    () -> job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}"));
+
+            job.setExecuteSql("INSERT INTO");
+            Assert.assertThrows(ParseException.class, () -> job.initLogicalPlan(true));
+            Assert.assertSame(caller, ConnectContext.get());
+        } finally {
+            ConnectContext.remove();
+            if (previousContext != null) {
+                previousContext.setThreadLocalInfo();
+            }
+        }
     }
 
     @Test

fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java

@@ -29,6 +29,10 @@ import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.datasource.jdbc.client.JdbcClient;
 import org.apache.doris.job.cdc.DataSourceConfigKeys;
 import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.GlobalVariable;
 
 import org.junit.Assert;
@@ -56,6 +60,32 @@ public class StreamingJobUtilsTest {
         MockitoAnnotations.initMocks(this);
     }
 
+    @Test
+    public void testCdcDriverPrivilege() {
+        Env env = Mockito.mock(Env.class);
+        ConnectContext ctx = Mockito.mock(ConnectContext.class);
+        AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class);
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+            Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+            List<String> remoteDrivers = Arrays.asList("http://example.invalid/driver.jar",
+                    "https://example.invalid/driver.jar");
+            for (String driver : remoteDrivers) {
+                AnalysisException error = Assert.assertThrows(AnalysisException.class,
+                        () -> StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver));
+                Assert.assertTrue(error.getMessage().contains("ADMIN"));
+            }
+            Mockito.when(accessManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(true);
+            for (String driver : remoteDrivers) {
+                StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver);
+            }
+            Mockito.when(accessManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(false);
+            for (String driver : Arrays.asList(null, "driver.jar", "file:///opt/doris/driver.jar")) {
+                StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver);
+            }
+        }
+    }
+
     @Test
     public void testBinarySourceColumnsUseSupportedOlapTypes() {
         List<Column> columns = new ArrayList<>();

regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy

@@ -100,14 +100,34 @@ suite("test_streaming_mysql_job_priv", "p0,external,mysql,external_docker,extern
             sql """INSERT INTO ${mysqlDb}.${tableName} (name, age) VALUES ('B1', 2);"""
         }
 
+        // Prepare a local driver without granting ADMIN to the job creator.
+        File driverFile = new File("${context.config.dataPath}/jdbc_driver/mysql_priv_driver.jar").absoluteFile
+        driverFile.parentFile.mkdirs()
+        def download = new ProcessBuilder("curl", "--fail", "--location", "--max-time", "120",
+                driver_url, "--output", driverFile.path).redirectErrorStream(true).start()
+        String downloadOutput = download.text
+        assertEquals(0, download.waitFor(), downloadOutput)
+        scp_udf_file_to_all_fe(driverFile.path)
+        String localDriverUrl = "file://${driverFile.path}"
+
         // Create the job without ALTER privilege to verify schema changes use the job creator's identity.
         sql """grant load_priv,create_priv on ${dbName}.* to ${user}"""
         connect(user, "${pwd}", url) {
+            test {
+                sql """CREATE JOB ${jobName} ON STREAMING
+                        FROM MYSQL (
+                            "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
+                            "driver_url" = "${driver_url}", "driver_class" = "com.mysql.cj.jdbc.Driver",
+                            "user" = "root", "password" = "123456", "database" = "${mysqlDb}",
+                            "include_tables" = "${tableName}", "offset" = "initial")
+                        TO DATABASE ${currentDb} ("table.create.properties.replication_num" = "1")"""
+                exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege"
+            }
             sql """CREATE JOB ${jobName}
                     ON STREAMING
                     FROM MYSQL (
                         "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}?allowPublicKeyRetrieval=true&useSSL=false",
-                        "driver_url" = "${driver_url}",
+                        "driver_url" = "${localDriverUrl}",
                         "driver_class" = "com.mysql.cj.jdbc.Driver",
                         "user" = "root",
                         "password" = "123456",
@@ -191,6 +211,13 @@ suite("test_streaming_mysql_job_priv", "p0,external,mysql,external_docker,extern
         }
 
         sql "PAUSE JOB where jobname =  '${jobName}'";
+        connect(user, "${pwd}", url) {
+            test {
+                sql """ALTER JOB ${jobName} FROM MYSQL ("driver_url" = "${driver_url}")
+                        TO DATABASE ${currentDb}"""
+                exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege"
+            }
+        }
 
         sql """ALTER JOB ${jobName}
                 FROM MYSQL (

regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy

@@ -138,6 +138,27 @@ suite("test_cdc_stream_tvf_mysql", "p0,external,mysql,external_docker,external_d
             exception "ssl_mode 'verify-ca' requires ssl_rootcert to be set"
         }
 
+        // Remote driver URLs require ADMIN even for standalone TVF queries.
+        String driverUser = "test_cdc_stream_tvf_mysql_driver_user"
+        def jdbcTokens = context.getJdbcUrl().split('/')
+        String driverUserUrl = jdbcTokens[0] + "//" + jdbcTokens[2] + "/" + currentDb + "?"
+        sql """DROP USER IF EXISTS '${driverUser}'"""
+        sql """CREATE USER '${driverUser}' IDENTIFIED BY '123456'"""
+        sql """GRANT SELECT_PRIV ON ${currentDb}.* TO '${driverUser}'"""
+        try {
+            connectToDoris(driverUser, "123456", driverUserUrl) {
+                test {
+                    sql """SELECT * FROM cdc_stream(
+                            "type" = "mysql", "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
+                            "database" = "${mysqlDb}", "table" = "${table1}", "offset" = "latest",
+                            "driver_url" = "${driver_url}")"""
+                    exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege"
+                }
+            }
+        } finally {
+            sql """DROP USER '${driverUser}'"""
+        }
+
         // --- Data setup ---
 
         connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") {

regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy

@@ -67,10 +67,8 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker,
             sql """INSERT INTO ${mysqlDb}.${mysqlTable} (name, age) VALUES ('B1', 2)"""
         }
 
-        // create streaming job via cdc_stream TVF (offset=initial → snapshot then binlog)
-        sql """
-            CREATE JOB ${jobName}
-            ON STREAMING DO INSERT INTO ${currentDb}.${dorisTable} (name, age)
+        String insertSql = """
+            INSERT INTO ${currentDb}.${dorisTable} (name, age)
             SELECT name, age FROM cdc_stream(
                 "type"         = "mysql",
                 "jdbc_url"     = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
@@ -85,6 +83,25 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker,
             )
         """
 
+        String driverUser = "test_streaming_cdc_mysql_driver_user"
+        def jdbcTokens = context.getJdbcUrl().split('/')
+        String driverUserUrl = jdbcTokens[0] + "//" + jdbcTokens[2] + "/" + currentDb + "?"
+        sql """DROP USER IF EXISTS '${driverUser}'"""
+        sql """CREATE USER '${driverUser}' IDENTIFIED BY '123456'"""
+        sql """GRANT LOAD_PRIV ON ${currentDb}.* TO '${driverUser}'"""
+        if (isCloudMode()) {
+            def cluster = sql_return_maparray("SHOW CLUSTERS").find { it.is_current.equalsIgnoreCase("TRUE") }.cluster
+            sql """GRANT USAGE_PRIV ON CLUSTER `${cluster}` TO '${driverUser}'"""
+        }
+        connectToDoris(driverUser, "123456", driverUserUrl) {
+            test {
+                sql """CREATE JOB ${jobName} ON STREAMING DO ${insertSql}"""
+                exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege"
+            }
+        }
+        // The administrator can create the same job with the remote driver.
+        sql """CREATE JOB ${jobName} ON STREAMING DO ${insertSql}"""
+
         // wait for at least one snapshot task to succeed
         try {
             Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({
@@ -130,6 +147,33 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker,
             s.size() == 1 && s.get(0).get(0) == "PAUSED"
         })
 
+        String driverResource = "test_streaming_cdc_mysql_driver_resource"
+        try_sql """DROP RESOURCE '${driverResource}'"""
+        try {
+            sql """CREATE RESOURCE '${driverResource}' PROPERTIES (
+                "type" = "jdbc",
+                "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
+                "user" = "root", "password" = "123456",
+                "driver_url" = "${driver_url}", "driver_class" = "com.mysql.cj.jdbc.Driver"
+            )"""
+            // ALTER does not read the local jar; the conflicting SQL must be rejected first.
+            sql """ALTER RESOURCE '${driverResource}' PROPERTIES ("driver_url" = "file:///trusted/mysql.jar")"""
+            sql """GRANT USAGE_PRIV ON RESOURCE '${driverResource}' TO '${driverUser}'"""
+            String inheritedDriverSql = insertSql.replace('"driver_url"',
+                    '"resource" = "' + driverResource + '", "Driver_url"')
+                    .replace(driver_url, 'https://example.invalid/driver.jar')
+                    .replace('com.mysql.cj.jdbc.Driver', 'example.InvalidDriver')
+            connectToDoris(driverUser, "123456", driverUserUrl) {
+                test {
+                    sql """ALTER JOB ${jobName} ${inheritedDriverSql}"""
+                    exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege"
+                }
+            }
+        } finally {
+            sql """DROP RESOURCE '${driverResource}'"""
+        }
+        sql """DROP USER '${driverUser}'"""
+
         test {
             sql """
                 ALTER JOB ${jobName}