This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 67eef18c2d8 [fix](load) Enforce auth for group commit stream load
(#69516) (#69747)
67eef18c2d8 is described below
commit 67eef18c2d8be6af55b5d16b921ca0e698383239
Author: Gavin Chou <[email protected]>
AuthorDate: Fri Oct 30 19:03:29 2026 +0800
[fix](load) Enforce auth for group commit stream load (#69516) (#69747)
## Summary
Backport #69516 to `branch-4.2`.
Explicit group commit Stream Load skips the normal transaction begin
path, so this applies the same credential and target-table LOAD
privilege checks before follower forwarding or local planning.
The target branch still uses JUnit 4; the test-only conflict was
resolved by adapting the upstream assertions/imports to JUnit 4.
Production behavior is unchanged from #69516.
## Testing
- [x] `sh run-fe-ut.sh --run
"org.apache.doris.service.FrontendServiceImplTest"` (25 tests, 0
failures/errors)
- [ ] `sh build.sh --fe` (the configured prebuilt package lacked the
required Paimon artifacts and triggered a local third-party rebuild;
configuration later stopped on an external `opencensus-proto v0.3.0`
download with `SSL connect error`, before FE compilation)
---------
Co-authored-by: gavinchou <[email protected]>
---
.../apache/doris/service/FrontendServiceImpl.java | 37 +++++++++++---
.../doris/service/FrontendServiceImplTest.java | 58 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 42599d5f9e1..0c751482760 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -2265,6 +2265,26 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return Math.floorMod(GROUP_COMMIT_FOLLOWER_INDEX.getAndIncrement(),
followerCount);
}
+ private void checkGroupCommitStreamLoadAuth(TStreamLoadPutRequest request)
throws AuthenticationException {
+ // Explicit group commit skips loadTxnBegin, so authenticate and
authorize the standard Stream Load
+ // request here before it can be forwarded or planned. Requests with
load_sql use the separate HTTP
+ // Stream path, whose planner performs its own authorization and whose
internal caller uses a token.
+ if (!Strings.isNullOrEmpty(request.getLoadSql()) ||
!request.isSetGroupCommitMode()
+ || request.getGroupCommitMode().equalsIgnoreCase("off_mode")) {
+ return;
+ }
+
+ if (request.isSetAuthCode()) {
+ // TODO: deprecated, removed in 3.1, use token instead.
+ } else if (Strings.isNullOrEmpty(request.getToken())) {
+ checkSingleTablePasswordAndPrivs(request.getUser(),
request.getPasswd(), request.getDb(),
+ request.getTbl(), request.getUserIp(), PrivPredicate.LOAD,
+ toForwardedCertificateInfo(request.getCertBasedAuth()));
+ } else {
+ checkTokenOrThrow(request.getToken());
+ }
+ }
+
private TStreamLoadPutResult
forwardGroupCommitStreamLoad(TStreamLoadPutRequest request) {
HostInfo selfNode = Env.getCurrentEnv().getSelfNode();
List<Frontend> followers =
Env.getCurrentEnv().getFrontends(FrontendNodeType.FOLLOWER).stream()
@@ -2334,14 +2354,6 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
}
String groupCommitMode = request.getGroupCommitMode();
- if (groupCommitMode != null && !groupCommitMode.equals("off_mode") &&
Env.getCurrentEnv().isMaster()
- && Config.enable_forward_group_commit_stream_load_to_follower)
{
- TStreamLoadPutResult result =
forwardGroupCommitStreamLoad(request);
- if (result != null) {
- return result;
- }
- }
-
TStreamLoadPutResult result = new TStreamLoadPutResult();
TStatus status = new TStatus(TStatusCode.OK);
result.setStatus(status);
@@ -2349,6 +2361,15 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
StreamLoadHandler streamLoadHandler = new StreamLoadHandler(request,
null, result, clientAddr);
try {
+ checkGroupCommitStreamLoadAuth(request);
+ if (groupCommitMode != null && !groupCommitMode.equals("off_mode")
&& Env.getCurrentEnv().isMaster()
+ &&
Config.enable_forward_group_commit_stream_load_to_follower) {
+ TStreamLoadPutResult forwardedResult =
forwardGroupCommitStreamLoad(request);
+ if (forwardedResult != null) {
+ return forwardedResult;
+ }
+ }
+
streamLoadHandler.setCloudCluster();
List<TPipelineWorkloadGroup> tWorkloadGroupList = null;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
index 6a4071c08c6..b97c34058f6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
@@ -73,7 +73,10 @@ import org.apache.doris.thrift.TSchemaTableRequestParams;
import org.apache.doris.thrift.TShowUserRequest;
import org.apache.doris.thrift.TShowUserResult;
import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TStreamLoadPutRequest;
+import org.apache.doris.thrift.TStreamLoadPutResult;
import org.apache.doris.thrift.TTableStatus;
+import org.apache.doris.thrift.TUniqueId;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.utframe.UtFrameUtils;
@@ -953,6 +956,61 @@ public class FrontendServiceImplTest {
}
}
+ @Test
+ public void
testGroupCommitStreamLoadPutChecksAuthenticationAndLoadPrivilege() throws
Exception {
+ String noLoadUser = "group_commit_no_load_user";
+ String noLoadPassword = "group_commit_test_password";
+ boolean originalForwardSetting =
Config.enable_forward_group_commit_stream_load_to_follower;
+ Config.enable_forward_group_commit_stream_load_to_follower = false;
+ executeCommand("DROP USER IF EXISTS " + noLoadUser);
+ executeCommand("CREATE USER " + noLoadUser + " IDENTIFIED BY '" +
noLoadPassword + "'");
+
+ try {
+ FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+
+ TStreamLoadPutResult badPassword = impl.streamLoadPut(
+ groupCommitStreamLoadRequest("root", "wrong_password",
"missing_table"));
+ Assert.assertEquals(TStatusCode.ANALYSIS_ERROR,
badPassword.getStatus().getStatusCode());
+
Assert.assertTrue(badPassword.getStatus().getErrorMsgs().get(0).contains("Access
denied"));
+
+ TStreamLoadPutResult missingLoad = impl.streamLoadPut(
+ groupCommitStreamLoadRequest(noLoadUser, noLoadPassword,
"missing_table"));
+ Assert.assertEquals(TStatusCode.ANALYSIS_ERROR,
missingLoad.getStatus().getStatusCode());
+ String missingLoadError =
missingLoad.getStatus().getErrorMsgs().get(0);
+ Assert.assertTrue(missingLoadError,
missingLoadError.contains("Load_priv"));
+
+ FrontendServiceImpl tokenCheckingImpl = Mockito.spy(new
FrontendServiceImpl(exeEnv));
+ TStreamLoadPutRequest badTokenRequest =
groupCommitStreamLoadRequest("", "", "missing_table");
+ badTokenRequest.setToken("bad-token");
+
Mockito.doReturn(false).when(tokenCheckingImpl).checkToken("bad-token");
+ TStreamLoadPutResult badToken =
tokenCheckingImpl.streamLoadPut(badTokenRequest);
+ Assert.assertEquals(TStatusCode.ANALYSIS_ERROR,
badToken.getStatus().getStatusCode());
+
Assert.assertTrue(badToken.getStatus().getErrorMsgs().get(0).contains("Invalid
token"));
+
+ TStreamLoadPutResult authorized = impl.streamLoadPut(
+ groupCommitStreamLoadRequest("root", "", "missing_table"));
+ TStatusCode authorizedStatus =
authorized.getStatus().getStatusCode();
+ Assert.assertTrue("authorized request should reach the
missing-table check, but got " + authorizedStatus,
+ authorizedStatus == TStatusCode.NOT_FOUND ||
authorizedStatus == TStatusCode.ANALYSIS_ERROR);
+
Assert.assertTrue(authorized.getStatus().getErrorMsgs().get(0).contains("missing_table"));
+ } finally {
+ connectContext.setThreadLocalInfo();
+ executeCommand("DROP USER IF EXISTS " + noLoadUser);
+ Config.enable_forward_group_commit_stream_load_to_follower =
originalForwardSetting;
+ }
+ }
+
+ private static TStreamLoadPutRequest groupCommitStreamLoadRequest(String
user, String password, String table) {
+ return new TStreamLoadPutRequest()
+ .setUser(user)
+ .setPasswd(password)
+ .setDb("test")
+ .setTbl(table)
+ .setUserIp("127.0.0.1")
+ .setLoadId(new TUniqueId(1L, 1L))
+ .setGroupCommitMode("async_mode");
+ }
+
private MockedStatic<Env> transactionValidationEnvMock;
private void mockTransactionForTokenValidation(long txnId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
