mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
[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.
This commit is contained in:
9 files changed
+208
-18
No files matched your search
+19
-3
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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());
|
||||
|
||||
+5
-1
@@ -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());
|
||||
|
||||
+3
@@ -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");
|
||||
|
||||
+42
-9
@@ -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();
|
||||
|
||||
Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus());
|
||||
Assertions.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\"}");
|
||||
|
||||
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
|
||||
|
||||
@@ -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<>();
|
||||
|
||||
+28
-1
@@ -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 (
|
||||
|
||||
@@ -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}") {
|
||||
|
||||
+48
-4
@@ -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}
|
||||
|
||||
Reference in new issue
Block a user