This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c8a75b273e9 [fix](fe) Require ADMIN for remote CDC driver URLs (#69617)
c8a75b273e9 is described below
commit c8a75b273e999370e354ed39ae65db2c483c0ad0
Author: wudi <[email protected]>
AuthorDate: Tue Nov 3 16:10:48 2026 +0800
[fix](fe) Require ADMIN for remote CDC driver URLs (#69617)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
Users with permission to create CDC jobs can supply HTTP/HTTPS JDBC
driver URLs that load code into the FE JVM. Require global ADMIN
privilege when these URLs are supplied during CDC job creation,
alteration, or `cdc_stream` construction, before source initialization
or driver loading.
Share the permission check across the FROM-TO and TVF paths. Preserve
existing local driver permissions and S3 streaming behavior.
---
.../insert/streaming/StreamingInsertJob.java | 22 +++++++--
.../apache/doris/job/util/StreamingJobUtils.java | 12 +++++
.../trees/plans/commands/AlterJobCommand.java | 6 ++-
.../CdcStreamTableValuedFunction.java | 3 ++
.../StreamingInsertJobOffsetPersistenceTest.java | 51 +++++++++++++++++----
.../doris/job/util/StreamingJobUtilsTest.java | 30 +++++++++++++
.../cdc/test_streaming_mysql_job_priv.groovy | 29 +++++++++++-
.../cdc/tvf/test_cdc_stream_tvf_mysql.groovy | 21 +++++++++
.../tvf/test_streaming_job_cdc_stream_mysql.groovy | 52 ++++++++++++++++++++--
9 files changed, 208 insertions(+), 18 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index add7761ddc2..a1ee8ccc85a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -79,6 +79,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;
@@ -267,6 +268,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();
@@ -429,9 +432,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.
@@ -489,9 +496,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();
+ }
+ }
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
index 5c5b4064338..88a98854e49 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
@@ -31,6 +31,7 @@ import org.apache.doris.catalog.info.TableNameInfo;
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());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
index efa1757a6d5..965989002f6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
+++
b/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());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
index 0a53ec38479..14f02c80a62 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
+++
b/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");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
index 63bce5df853..c07703f4b2c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
+++
b/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;
@@ -121,21 +126,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");
Assertions.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();
+
+ // Restored jobs have no cached plan/TVF properties. Parsing must
not replace the caller.
+ job.validateAlterOffset("{\"lsn\":\"300\"}");
- Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus());
- Assertions.assertThrows(AnalysisException.class,
- () ->
job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}"));
+ Assertions.assertSame(caller, ConnectContext.get());
+ Assertions.assertEquals(Collections.singleton("session_role"),
+ ConnectContext.get().getAuthenticatedRoles());
+ Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus());
+ Assertions.assertThrows(AnalysisException.class,
+ () ->
job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}"));
+
+ job.setExecuteSql("INSERT INTO");
+ Assertions.assertThrows(ParseException.class, () ->
job.initLogicalPlan(true));
+ Assertions.assertSame(caller, ConnectContext.get());
+ } finally {
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
index b55fa2c817c..ee72dd9c043 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
@@ -30,6 +30,10 @@ 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.job.exception.JobException;
+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.jupiter.api.Assertions;
@@ -57,6 +61,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 =
List.of("http://example.invalid/driver.jar";,
+ "https://example.invalid/driver.jar";);
+ for (String driver : remoteDrivers) {
+ AnalysisException error =
Assertions.assertThrows(AnalysisException.class,
+ () -> StreamingJobUtils.checkCdcDriverPrivilege(ctx,
driver));
+ Assertions.assertTrue(error.getMessage().contains("ADMIN"));
+ }
+ Mockito.when(accessManager.checkGlobalPriv(ctx,
PrivPredicate.ADMIN)).thenReturn(true);
+ for (String driver : remoteDrivers) {
+ Assertions.assertDoesNotThrow(() ->
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")) {
+ Assertions.assertDoesNotThrow(() ->
StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver));
+ }
+ }
+ }
+
@Test
public void testBinarySourceColumnsUseSupportedOlapTypes() {
List<Column> columns = new ArrayList<>();
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy
index 6100c7eee9e..1507d3c3b5f 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy
+++
b/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 (
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
index 744ab7539ac..c9b6b766e82 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
+++
b/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}") {
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy
index d0848257a8b..78d8b76a897 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy
+++
b/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}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
