github-actions[bot] commented on code in PR #65607:
URL: https://github.com/apache/doris/pull/65607#discussion_r3580471996
##########
fe/fe-core/src/main/java/org/apache/doris/DorisFE.java:
##########
@@ -136,16 +136,23 @@ public static void start(String dorisHomeDir, String
pidDir, String[] args, Star
return;
}
- CommandLineOptions cmdLineOpts = parseArgs(args);
-
try {
+ CommandLine commandLine = parseArgs(args);
+ if (commandLine.hasOption('v') ||
commandLine.hasOption("version")) {
+ printVersion();
Review Comment:
This now calls `printVersion()` before `Config.init()`, but `printVersion()`
still emits the run-mode line from `Config.isCloudMode()`. In a fresh JVM those
fields still have their defaults (`deploy_mode = ""`, `cloud_unique_id = ""`),
so a cloud FE started with `--version` will print `Run FE in the local mode`
instead of the configured cloud mode. Either keep the early path to
config-independent build fields only, or load enough config before printing the
mode line; the current test only checks that missing config does not fail, not
that the output is still correct.
##########
fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java:
##########
@@ -60,8 +63,12 @@ public class GroupCommitManager {
private Set<Long> blockedTableIds = new HashSet<>();
- // Encoded <Cluster and Table id> to BE id map. Only for group commit.
- private final Map<String, Long> tableToBeMap = new ConcurrentHashMap<>();
+ // Encoded <Cluster and Table id> to BE id cache. Only for group commit.
Review Comment:
This still leaves the other half of the per-table group-commit state
unbounded. Each cache miss that selects a BE creates
`tableToPressureMap.put(tableId, new SlidingWindowCounter(...))`, and
`updateLoadDataInternal()` only ever updates that entry; I could not find any
`tableToPressureMap.remove`, expiry, or drop-table cleanup path. So after this
patch, a workload that group-commit-loads and drops many tables will evict the
BE id from `tableToBeMap`, but will still keep one pressure counter per table
id forever. Please give the pressure map the same bounded/expiry lifecycle, or
explicitly remove both structures when table group-commit state is no longer
valid.
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -1152,90 +1152,131 @@ public TFetchResourceResult fetchResource() throws
TException {
@Override
public TMasterOpResult forward(TMasterOpRequest params) throws TException {
+ validateForwardRequester(params);
+ TMasterOpResult shortcut = handleForwardShortcut(params);
+ if (shortcut != null) {
+ return shortcut;
+ }
+ logForwardRequest(params);
+ ConnectContext context = createForwardContext(params);
+ ConnectProcessor processor = createForwardProcessor(context);
+ Runnable clearCallback = registerProxyQuery(params, context);
+ try {
+ return executeForward(params, context, processor);
+ } finally {
+ ConnectContext.remove();
+ clearCallback.run();
+ }
+ }
+
+ private void validateForwardRequester(TMasterOpRequest params) throws
TException {
Frontend fe =
Env.getCurrentEnv().checkFeExist(params.getClientNodeHost(),
params.getClientNodePort());
- if (fe == null) {
- LOG.warn("reject request from invalid host. client: {}",
params.getClientNodeHost());
- throw new TException("request from invalid host was rejected.");
+ if (fe != null) {
+ return;
}
+ LOG.warn("reject request from invalid host. client: {}",
params.getClientNodeHost());
+ throw new TException("request from invalid host was rejected.");
+ }
+
+ private TMasterOpResult handleForwardShortcut(TMasterOpRequest params)
throws TException {
if (params.isSyncJournalOnly()) {
- final TMasterOpResult result = new TMasterOpResult();
- result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ return createForwardResultWithJournalSync();
}
if (params.getGroupCommitInfo() != null &&
params.getGroupCommitInfo().isGetGroupCommitLoadBeId()) {
- final TGroupCommitInfo info = params.getGroupCommitInfo();
- final TMasterOpResult result = new TMasterOpResult();
- try {
-
result.setGroupCommitLoadBeId(Env.getCurrentEnv().getGroupCommitManager()
-
.selectBackendForGroupCommitInternal(info.groupCommitLoadTableId,
info.cluster));
- } catch (LoadException | DdlException e) {
- throw new TException(e.getMessage());
- }
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ return handleGroupCommitLoadBeId(params.getGroupCommitInfo());
}
if (params.getGroupCommitInfo() != null &&
params.getGroupCommitInfo().isUpdateLoadData()) {
- final TGroupCommitInfo info = params.getGroupCommitInfo();
- final TMasterOpResult result = new TMasterOpResult();
Env.getCurrentEnv().getGroupCommitManager()
- .updateLoadData(info.tableId, info.receiveData);
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ .updateLoadData(params.getGroupCommitInfo().tableId,
params.getGroupCommitInfo().receiveData);
+ return createForwardResultWithoutJournalSync();
}
- if (params.isSetCancelQeury() && params.isCancelQeury()) {
- if (!params.isSetQueryId()) {
- throw new TException("a query id is needed to cancel a query");
- }
- TUniqueId queryId = params.getQueryId();
- ConnectContext ctx = proxyQueryIdToConnCtx.get(queryId);
- if (ctx != null) {
- ctx.cancelQuery(new Status(TStatusCode.CANCELLED, "cancel
query by forward request."));
- }
- final TMasterOpResult result = new TMasterOpResult();
- result.setStatusCode(0);
- result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
- // just make the protocol happy
- result.setPacket("".getBytes());
- return result;
+ if (!params.isSetCancelQeury() || !params.isCancelQeury()) {
+ return null;
+ }
+ return handleForwardCancel(params);
+ }
+
+ private TMasterOpResult createForwardResultWithJournalSync() {
+ TMasterOpResult result = new TMasterOpResult();
+ result.setMaxJournalId(Env.getCurrentEnv().getMaxJournalId());
+ result.setPacket("".getBytes());
+ return result;
+ }
+
+ private TMasterOpResult createForwardResultWithoutJournalSync() {
Review Comment:
Even though these shortcuts do not need journal synchronization, this helper
still returns a `TMasterOpResult`, whose IDL declares `maxJournalId` as a
required field. The new helper only sets `packet`, so both group-commit
shortcut responses now centralize an incomplete result struct. Please set an
explicit no-sync value such as `result.setMaxJournalId(0L)` here, or make the
field optional through an intentional compatibility change, so the shortcut
response still satisfies the thrift contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
