Add REST write request size and batch limits

external-service-impl/rest/src/main/i18n/en/org/apache/iotdb/rest/i18n/RestMessages.java

@@ -77,6 +77,49 @@ public final class RestMessages {
       "The number of values in the %dth row is not equal to the data_types size";
   public static final String ERROR_MESSAGE_SEPARATOR = ",";
 
+  // --- RequestSizeLimitFilter ---
+  public static final String
+      MESSAGE_REST_REQUEST_BODY_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_424392C6 =
+          "REST request body exceeds limit %d bytes. Use SET CONFIGURATION"
+              + " 'rest_max_request_body_size_in_bytes'='<bytes>' to increase it.";
+  public static final String
+      MESSAGE_REST_REQUEST_BODY_MEMORY_QUOTA_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_TOTAL_CONCURRENT_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_F07B9DDD =
+          "REST request body memory quota exceeds limit %d bytes. Use SET CONFIGURATION"
+              + " 'rest_max_total_concurrent_request_body_size_in_bytes'='<bytes>' to increase it.";
+
+  // --- RequestLimitChecker ---
+  public static final String MESSAGE_INSERTTABLET_REQUEST_8647CA58 = "insertTablet request";
+  public static final String MESSAGE_INSERTRECORDS_REQUEST_93E12369 = "insertRecords request";
+  public static final String MESSAGE_TABLE_INSERTTABLET_REQUEST_573D371C =
+      "table insertTablet request";
+  public static final String EXCEPTION_ARG_ROW_COUNT_ARG_EXCEEDS_LIMIT_ARG_EE427E4B =
+      "%s row count %d exceeds limit %d";
+  public static final String EXCEPTION_ARG_COLUMN_COUNT_ARG_EXCEEDS_LIMIT_ARG_DEE9637E =
+      "%s column count %d exceeds limit %d";
+  public static final String EXCEPTION_ARG_VALUE_COUNT_ARG_EXCEEDS_LIMIT_ARG_77F95703 =
+      "%s value count %d exceeds limit %d";
+
+  // --- RequestValidationHandler ---
+  public static final String
+      EXCEPTION_MEASUREMENTS_AND_DATATYPES_SHOULD_HAVE_THE_SAME_SIZE_FF715FA9 =
+          "measurements and dataTypes should have the same size";
+  public static final String EXCEPTION_VALUES_AND_DATATYPES_SHOULD_HAVE_THE_SAME_SIZE_5BC1D604 =
+      "values and dataTypes should have the same size";
+  public static final String
+      EXCEPTION_EACH_VALUE_COLUMN_SHOULD_HAVE_THE_SAME_SIZE_AS_TIMESTAMPS_523598BD =
+          "Each value column should have the same size as timestamps";
+  public static final String
+      EXCEPTION_MEASUREMENTS_AND_DATA_TYPES_SHOULD_HAVE_THE_SAME_SIZE_8526F19A =
+          "measurements and data_types should have the same size";
+  public static final String EXCEPTION_VALUES_AND_DATA_TYPES_SHOULD_HAVE_THE_SAME_SIZE_0BAE701D =
+      "values and data_types should have the same size";
+  public static final String
+      EXCEPTION_DEVICES_TIMESTAMPS_MEASUREMENTS_LIST_DATA_TYPES_LIST_AND_VALUES_LIST_SHOULD_HAVE_THE_SAME_SIZE_5983AAC2 =
+          "devices, timestamps, measurements_list, data_types_list and values_list should have the same size";
+  public static final String
+      EXCEPTION_EACH_INSERTRECORDS_ROW_SHOULD_HAVE_THE_SAME_NUMBER_OF_MEASUREMENTS_DATA_TYPES_AND_VALUES_AD58AEF2 =
+          "Each insertRecords row should have the same number of measurements, data types and values";
+
   private RestMessages() {}
   // ---------------------------------------------------------------------------
   // Additional auto-collected messages

external-service-impl/rest/src/main/i18n/zh/org/apache/iotdb/rest/i18n/RestMessages.java

@@ -75,6 +75,49 @@ public final class RestMessages {
       "第 %d 行的 values 数量与 data_types 数量不相等";
   public static final String ERROR_MESSAGE_SEPARATOR = "，";
 
+  // --- RequestSizeLimitFilter ---
+  public static final String
+      MESSAGE_REST_REQUEST_BODY_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_424392C6 =
+          "REST 请求体超过限制 %d 字节。请执行 SET CONFIGURATION"
+              + " 'rest_max_request_body_size_in_bytes'='<bytes>' 增大限制。";
+  public static final String
+      MESSAGE_REST_REQUEST_BODY_MEMORY_QUOTA_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_TOTAL_CONCURRENT_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_F07B9DDD =
+          "REST 请求体内存配额超过限制 %d 字节。请执行 SET CONFIGURATION"
+              + " 'rest_max_total_concurrent_request_body_size_in_bytes'='<bytes>' 增大配额。";
+
+  // --- RequestLimitChecker ---
+  public static final String MESSAGE_INSERTTABLET_REQUEST_8647CA58 = "insertTablet 请求";
+  public static final String MESSAGE_INSERTRECORDS_REQUEST_93E12369 = "insertRecords 请求";
+  public static final String MESSAGE_TABLE_INSERTTABLET_REQUEST_573D371C =
+      "table insertTablet 请求";
+  public static final String EXCEPTION_ARG_ROW_COUNT_ARG_EXCEEDS_LIMIT_ARG_EE427E4B =
+      "%s 行数 %d 超过限制 %d";
+  public static final String EXCEPTION_ARG_COLUMN_COUNT_ARG_EXCEEDS_LIMIT_ARG_DEE9637E =
+      "%s 列数 %d 超过限制 %d";
+  public static final String EXCEPTION_ARG_VALUE_COUNT_ARG_EXCEEDS_LIMIT_ARG_77F95703 =
+      "%s value 数 %d 超过限制 %d";
+
+  // --- RequestValidationHandler ---
+  public static final String
+      EXCEPTION_MEASUREMENTS_AND_DATATYPES_SHOULD_HAVE_THE_SAME_SIZE_FF715FA9 =
+          "measurements 和 dataTypes 的数量应相同";
+  public static final String EXCEPTION_VALUES_AND_DATATYPES_SHOULD_HAVE_THE_SAME_SIZE_5BC1D604 =
+      "values 和 dataTypes 的数量应相同";
+  public static final String
+      EXCEPTION_EACH_VALUE_COLUMN_SHOULD_HAVE_THE_SAME_SIZE_AS_TIMESTAMPS_523598BD =
+          "每个 value 列的数量应与 timestamps 数量相同";
+  public static final String
+      EXCEPTION_MEASUREMENTS_AND_DATA_TYPES_SHOULD_HAVE_THE_SAME_SIZE_8526F19A =
+          "measurements 和 data_types 的数量应相同";
+  public static final String EXCEPTION_VALUES_AND_DATA_TYPES_SHOULD_HAVE_THE_SAME_SIZE_0BAE701D =
+      "values 和 data_types 的数量应相同";
+  public static final String
+      EXCEPTION_DEVICES_TIMESTAMPS_MEASUREMENTS_LIST_DATA_TYPES_LIST_AND_VALUES_LIST_SHOULD_HAVE_THE_SAME_SIZE_5983AAC2 =
+          "devices、timestamps、measurements_list、data_types_list 和 values_list 的数量应相同";
+  public static final String
+      EXCEPTION_EACH_INSERTRECORDS_ROW_SHOULD_HAVE_THE_SAME_NUMBER_OF_MEASUREMENTS_DATA_TYPES_AND_VALUES_AD58AEF2 =
+          "每个 insertRecords 行中的 measurements、data types 和 values 数量应相同";
+
   private RestMessages() {}
   // ---------------------------------------------------------------------------
   // Additional auto-collected messages

external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/exception/RequestLimitExceededException.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.rest.protocol.exception;
+
+public class RequestLimitExceededException extends IllegalArgumentException {
+
+  public RequestLimitExceededException(String message) {
+    super(message);
+  }
+}

external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/filter/RequestBodyMemoryReleaseFilter.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.rest.protocol.filter;
+
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerResponseContext;
+import jakarta.ws.rs.container.ContainerResponseFilter;
+import jakarta.ws.rs.ext.Provider;
+
+@Provider
+public class RequestBodyMemoryReleaseFilter implements ContainerResponseFilter {
+
+  @Override
+  public void filter(
+      ContainerRequestContext requestContext, ContainerResponseContext responseContext) {
+    RestRequestBodyMemoryManager.releaseReservation(requestContext);
+  }
+}

external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/filter/RequestSizeLimitFilter.java

@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.rest.protocol.filter;
+
+import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor;
+import org.apache.iotdb.rest.i18n.RestMessages;
+import org.apache.iotdb.rest.protocol.model.ExecutionStatus;
+
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.container.PreMatching;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.ext.Provider;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+@Provider
+@PreMatching
+public class RequestSizeLimitFilter implements ContainerRequestFilter {
+
+  private static final int PAYLOAD_TOO_LARGE_STATUS_CODE = 413;
+  private static final int SERVICE_UNAVAILABLE_STATUS_CODE = 503;
+
+  @Override
+  public void filter(ContainerRequestContext requestContext) {
+    long maxBodySize =
+        IoTDBRestServiceDescriptor.getInstance().getConfig().getRestMaxRequestBodySizeInBytes();
+    long memoryLimit =
+        IoTDBRestServiceDescriptor.getInstance()
+            .getConfig()
+            .getRestMaxTotalConcurrentRequestBodySizeInBytes();
+    if (maxBodySize <= 0 && memoryLimit <= 0) {
+      return;
+    }
+
+    int contentLength = requestContext.getLength();
+    if (maxBodySize > 0 && contentLength > maxBodySize) {
+      requestContext.abortWith(buildPayloadTooLargeResponse(maxBodySize));
+      return;
+    }
+
+    RestRequestBodyMemoryManager.Reservation memoryReservation =
+        RestRequestBodyMemoryManager.newReservation(memoryLimit);
+    long memoryReservedByContentLength = 0;
+    if (contentLength > 0 && memoryReservation.isEnabled()) {
+      if (!memoryReservation.reserve(contentLength)) {
+        memoryReservation.close();
+        requestContext.abortWith(buildMemoryQuotaExceededResponse(memoryLimit));
+        return;
+      }
+      memoryReservedByContentLength = contentLength;
+      RestRequestBodyMemoryManager.registerReservation(requestContext, memoryReservation);
+    }
+
+    requestContext.setEntityStream(
+        new LimitedInputStream(
+            requestContext.getEntityStream(),
+            maxBodySize,
+            memoryLimit,
+            memoryReservation,
+            memoryReservedByContentLength));
+    if (memoryReservation.isEnabled() && memoryReservedByContentLength == 0) {
+      RestRequestBodyMemoryManager.registerReservation(requestContext, memoryReservation);
+    }
+  }
+
+  private static Response buildPayloadTooLargeResponse(long maxBodySize) {
+    return Response.status(PAYLOAD_TOO_LARGE_STATUS_CODE)
+        .type(MediaType.APPLICATION_JSON_TYPE)
+        .entity(
+            new ExecutionStatus()
+                .code(PAYLOAD_TOO_LARGE_STATUS_CODE)
+                .message(
+                    String.format(
+                        RestMessages
+                            .MESSAGE_REST_REQUEST_BODY_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_424392C6,
+                        maxBodySize)))
+        .build();
+  }
+
+  private static Response buildMemoryQuotaExceededResponse(long memoryLimit) {
+    return Response.status(SERVICE_UNAVAILABLE_STATUS_CODE)
+        .type(MediaType.APPLICATION_JSON_TYPE)
+        .entity(
+            new ExecutionStatus()
+                .code(SERVICE_UNAVAILABLE_STATUS_CODE)
+                .message(
+                    String.format(
+                        RestMessages
+                            .MESSAGE_REST_REQUEST_BODY_MEMORY_QUOTA_EXCEEDS_LIMIT_ARG_BYTES_USE_SET_CONFIGURATION_REST_MAX_TOTAL_CONCURRENT_REQUEST_BODY_SIZE_IN_BYTES_BYTES_TO_INCREASE_IT_F07B9DDD,
+                        memoryLimit)))
+        .build();
+  }
+
+  private static class LimitedInputStream extends FilterInputStream {
+
+    private final long maxBodySize;
+    private final long memoryLimit;
+    private final RestRequestBodyMemoryManager.Reservation memoryReservation;
+    private long memoryCoveredBytes;
+    private long bytesRead;
+
+    private LimitedInputStream(
+        InputStream in,
+        long maxBodySize,
+        long memoryLimit,
+        RestRequestBodyMemoryManager.Reservation memoryReservation,
+        long memoryCoveredBytes) {
+      super(in);
+      this.maxBodySize = maxBodySize;
+      this.memoryLimit = memoryLimit;
+      this.memoryReservation = memoryReservation;
+      this.memoryCoveredBytes = memoryCoveredBytes;
+    }
+
+    @Override
+    public int read() throws IOException {
+      int result = super.read();
+      if (result != -1) {
+        incrementBytesRead(1);
+      }
+      return result;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+      int result = super.read(b, off, len);
+      if (result > 0) {
+        incrementBytesRead(result);
+      }
+      return result;
+    }
+
+    private void incrementBytesRead(int increment) {
+      bytesRead += increment;
+      if (maxBodySize > 0 && bytesRead > maxBodySize) {
+        memoryReservation.close();
+        throw new WebApplicationException(buildPayloadTooLargeResponse(maxBodySize));
+      }
+      reserveMemoryIfNecessary();
+    }
+
+    private void reserveMemoryIfNecessary() {
+      if (bytesRead <= memoryCoveredBytes) {
+        return;
+      }
+      long sizeToReserve = bytesRead - memoryCoveredBytes;
+      if (!memoryReservation.reserve(sizeToReserve)) {
+        memoryReservation.close();
+        throw new WebApplicationException(buildMemoryQuotaExceededResponse(memoryLimit));
+      }
+      memoryCoveredBytes = bytesRead;
+    }
+
+    @Override
+    public void close() throws IOException {
+      try {
+        super.close();
+      } finally {
+        memoryReservation.close();
+      }
+    }
+  }
+}