This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new e96ce958bac [To dev/1.3] Fix appearance of dispatching FI failed
because of thrift frame is oversize (#17550)
e96ce958bac is described below
commit e96ce958bac13ffeb51057e84751394b70f371ea
Author: Weihao Li <[email protected]>
AuthorDate: Mon Apr 27 14:25:05 2026 +0800
[To dev/1.3] Fix appearance of dispatching FI failed because of thrift
frame is oversize (#17550)
---
.../rpc/TCompressedElasticFramedTransport.java | 1 +
.../apache/iotdb/rpc/TElasticFramedTransport.java | 11 +++++++++++
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
.../iotdb/db/protocol/thrift/OperationType.java | 3 ++-
.../scheduler/FragmentInstanceDispatcherImpl.java | 11 +++++++++++
.../apache/iotdb/db/utils/ErrorHandlingUtils.java | 21 ++++++++++++++++++++-
6 files changed, 46 insertions(+), 2 deletions(-)
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
index a3b4f38064a..8579f46a3cb 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
@@ -75,6 +75,7 @@ public abstract class TCompressedElasticFramedTransport
extends TElasticFramedTr
writeCompressBuffer.resizeIfNecessary(maxCompressedLength);
int compressedLength =
compress(writeBuffer.getBuffer(), 0, length,
writeCompressBuffer.getBuffer(), 0);
+ checkWriteFrameSize(compressedLength);
RpcStat.writeCompressedBytes.addAndGet(compressedLength);
TFramedTransport.encodeFrameSize(compressedLength, i32buf);
underlying.write(i32buf, 0, 4);
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
index 67948a19f04..d98914ce934 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
@@ -199,9 +199,20 @@ public class TElasticFramedTransport extends TTransport {
readBuffer.fill(underlying, size);
}
+ protected void checkWriteFrameSize(int size) throws TTransportException {
+ if (size <= thriftMaxFrameSize) {
+ return;
+ }
+ close();
+ throw new TTransportException(
+ TTransportException.CORRUPTED_DATA,
+ "Frame size (" + size + ") larger than protect max size (" +
thriftMaxFrameSize + ")!");
+ }
+
@Override
public void flush() throws TTransportException {
int length = writeBuffer.getPos();
+ checkWriteFrameSize(length);
TFramedTransport.encodeFrameSize(length, i32buf);
underlying.write(i32buf, 0, 4);
underlying.write(writeBuffer.getBuffer(), 0, length);
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index c2f76d29484..e2855ed0a50 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -47,6 +47,7 @@ public enum TSStatusCode {
INTERNAL_SERVER_ERROR(305),
DISPATCH_ERROR(306),
LICENSE_ERROR(307),
+ THRIFT_FRAME_OVERSIZE(308),
// Client,
REDIRECTION_RECOMMEND(400),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/OperationType.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/OperationType.java
index 065a568353f..5838dac25bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/OperationType.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/OperationType.java
@@ -50,7 +50,8 @@ public enum OperationType {
CREATE_SCHEMA_TEMPLATE("createSchemaTemplate"),
CHECK_AUTHORITY("checkAuthority"),
EXECUTE_NON_QUERY_PLAN("executeNonQueryPlan"),
- QUERY_LATENCY("queryLatency");
+ QUERY_LATENCY("queryLatency"),
+ DISPATCH_FRAGMENT_INSTANCE("dispatchFragmentInstance");
private final String name;
OperationType(String name) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
index 7bcdd1fb694..41d325a5761 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -57,6 +57,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.thrift.TException;
+import org.apache.thrift.transport.TTransportException;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,6 +72,7 @@ import java.util.concurrent.TimeUnit;
import static com.google.common.util.concurrent.Futures.immediateFuture;
import static
org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.DISPATCH_READ;
+import static
org.apache.iotdb.db.utils.ErrorHandlingUtils.onThriftFrameOversizeException;
public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher
{
@@ -518,6 +520,15 @@ public class FragmentInstanceDispatcherImpl implements
IFragInstanceDispatcher {
TSStatusCode.EXECUTE_STATEMENT_ERROR,
String.format("unknown read type [%s]",
instance.getType())));
}
+ } catch (TException e) {
+ Throwable rootCause = ExceptionUtils.getRootCause(e);
+ if (rootCause instanceof TTransportException
+ && ((TTransportException) rootCause).getType() ==
TTransportException.CORRUPTED_DATA) {
+ // Don't set DISPATCH_ERROR status to avoid retry if dispatch failed
because of thrift frame
+ // is oversize
+ throw new
FragmentInstanceDispatchException(onThriftFrameOversizeException(rootCause));
+ }
+ throw e;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
index 963cecceb10..765f19c31bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java
@@ -46,6 +46,7 @@ import java.util.Objects;
import java.util.concurrent.ExecutionException;
import static org.apache.iotdb.commons.utils.StatusUtils.needRetry;
+import static
org.apache.iotdb.db.protocol.thrift.OperationType.DISPATCH_FRAGMENT_INSTANCE;
public class ErrorHandlingUtils {
@@ -60,10 +61,11 @@ public class ErrorHandlingUtils {
"The read statement is not allowed in batch: ";
private static final String ERROR_OPERATION_LOG = "Status code: {},
operation: {} failed";
+ private static final String EXCEPTION_PATTERN = "[%s] Exception occurred: %s
failed. ";
public static TSStatus onNpeOrUnexpectedException(
Exception e, String operation, TSStatusCode statusCode) {
- String message = String.format("[%s] Exception occurred: %s failed. ",
statusCode, operation);
+ String message = String.format(EXCEPTION_PATTERN, statusCode, operation);
if (e instanceof IOException || e instanceof NullPointerException) {
LOGGER.error(ERROR_OPERATION_LOG, statusCode, operation, e);
} else {
@@ -87,6 +89,23 @@ public class ErrorHandlingUtils {
return onNpeOrUnexpectedException(e, operation.getName(), statusCode);
}
+ public static Throwable getRootCause(Throwable e) {
+ while (e.getCause() != null) {
+ e = e.getCause();
+ }
+ return e;
+ }
+
+ public static TSStatus onThriftFrameOversizeException(Throwable t) {
+ TSStatus status =
+ new
TSStatus(TSStatusCode.THRIFT_FRAME_OVERSIZE.getStatusCode()).setNeedRetry(false);
+ String message =
+ String.format(EXCEPTION_PATTERN, status, DISPATCH_FRAGMENT_INSTANCE)
+ + getRootCause(t).getMessage();
+ LOGGER.warn(message);
+ return status.setMessage(message);
+ }
+
public static TSStatus onQueryException(Exception e, String operation,
TSStatusCode statusCode) {
TSStatus status = tryCatchQueryException(e);
if (status != null) {
