deardeng commented on code in PR #65607:
URL: https://github.com/apache/doris/pull/65607#discussion_r3580594167
##########
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:
maxJournalId is a required thrift field — does not reproduce.
maxJournalId is required i64, a primitive, and the Thrift Java generator
cannot enforce required primitives. From the generated
TMasterOpResult.validate():
public void validate() throws org.apache.thrift.TException {
// check for required fields
// alas, we cannot check 'maxJournalId' because it's a primitive and you
chose the non-beans generator.
if (packet == null) {
throw new org.apache.thrift.protocol.TProtocolException("Required
field 'packet' was not present! ...");
}
Required primitives are always written, using their default value when
unset, so the field is present on the wire either way. This is also not new
behavior: on master both group-commit shortcuts already build a TMasterOpResult
that only sets groupCommitLoadBeId / packet and never sets maxJournalId, and
group commit works today. The refactor centralizes that construction but does
not change what goes on the wire.
That said, relying on a default value to express "no journal id" is not
readable, so the helper now sets it explicitly:
private TMasterOpResult createForwardResultWithoutJournalSync() {
TMasterOpResult result = new TMasterOpResult();
// Group commit shortcuts update master memory without producing a
journal id. Say so explicitly
// instead of relying on the default value of the required maxJournalId
field.
result.setMaxJournalId(0L);
result.setPacket("".getBytes());
return result;
}
No protocol or compatibility change is needed.
--
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]
