connect-http-base Add a new parameter json_field_missed_return_null to let the user control whether to report an error or empty padding.

Author: ocean-zhc

Diff for: docs/en/connector-v2/source/Http.md

@@ -71,6 +71,8 @@ They can be downloaded via install-plugin.sh or from the Maven central repositor
 | common-options              |         | No       | -       | Source plugin common parameters, please refer to [Source Common Options](../source-common-options.md) for details                                                                 |
 | keep_params_as_form         |    Boolean     | No       | false       | Whether the params are submitted according to the form, used for compatibility with legacy behaviors. When true, the value of the params parameter is submitted through the form. |
 | keep_page_param_as_http_param         |    Boolean     | No       | false       | Whether to set the paging parameters to params. For compatibility with legacy behaviors.                                                                                                                                  |
+| json_filed_missed_return_null         |    Boolean     | No       | false       | When the json field is missing, set true return null else error.
+
 
 ## How to Create a Http Data Synchronization Jobs
 

Diff for: seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpParameter.java

@@ -45,6 +45,7 @@ public class HttpParameter implements Serializable {
     protected boolean enableMultilines;
     protected int connectTimeoutMs;
     protected int socketTimeoutMs;
+    protected boolean jsonFiledMissedReturnNull;
 
     public void buildWithConfig(ReadonlyConfig pluginConfig) {
         // set url
@@ -90,5 +91,7 @@ public void buildWithConfig(ReadonlyConfig pluginConfig) {
         this.setEnableMultilines(pluginConfig.get(HttpSourceOptions.ENABLE_MULTI_LINES));
         this.setConnectTimeoutMs(pluginConfig.get(HttpSourceOptions.CONNECT_TIMEOUT_MS));
         this.setSocketTimeoutMs(pluginConfig.get(HttpSourceOptions.SOCKET_TIMEOUT_MS));
+        this.setJsonFiledMissedReturnNull(
+                pluginConfig.get(HttpSourceOptions.JSON_FILED_MISSED_RETURN_NULL));
     }
 }

Diff for: seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSourceOptions.java

@@ -117,4 +117,10 @@ public class HttpSourceOptions extends HttpCommonOptions {
                     .intType()
                     .defaultValue(DEFAULT_SOCKET_TIMEOUT_MS)
                     .withDescription("Socket timeout setting, default 60s.");
+
+    public static final Option<Boolean> JSON_FILED_MISSED_RETURN_NULL =
+            Options.key("json_filed_missed_return_null")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("When the json field is missing, return null");
 }

Diff for: seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java

@@ -262,25 +262,41 @@ private List<List<String>> decodeJSON(String data) {
             List<String> result = jsonReadContext.read(path);
             results.add(result);
         }
-        int maxLength = 0;
-        for (List<?> result : results) {
-            maxLength = Math.max(maxLength, result.size());
-        }
-        for (int i = 0; i < results.size(); i++) {
-            List<String> result = results.get(i);
-            if (result.size() < maxLength) {
-                log.warn(
-                        "Field [{}] with size ({}) is less than max size ({}), will be padded with null values. "
-                                + "This may happen when JSON paths return different numbers of elements.",
-                        jsonPaths[i].getPath(),
-                        result.size(),
-                        maxLength);
-                for (int j = result.size(); j < maxLength; j++) {
-                    result.add(null);
+        if (httpParameter.isJsonFiledMissedReturnNull()) {
+            int maxLength = 0;
+            for (List<?> result : results) {
+                maxLength = Math.max(maxLength, result.size());
+            }
+            for (int i = 0; i < results.size(); i++) {
+                List<String> result = results.get(i);
+                if (result.size() < maxLength) {
+                    log.warn(
+                            "Field [{}] with size ({}) is less than max size ({}), will be padded with null values. "
+                                    + "This may happen when JSON paths return different numbers of elements.",
+                            jsonPaths[i].getPath(),
+                            result.size(),
+                            maxLength);
+                    for (int j = result.size(); j < maxLength; j++) {
+                        result.add(null);
+                    }
+                }
+            }
+        } else {
+            for (int i = 1; i < results.size(); i++) {
+                List<?> result0 = results.get(0);
+                List<?> result = results.get(i);
+                if (result0.size() != result.size()) {
+                    throw new HttpConnectorException(
+                            HttpConnectorErrorCode.FIELD_DATA_IS_INCONSISTENT,
+                            String.format(
+                                    "[%s](%d) and [%s](%d) the number of parsing records is inconsistent.",
+                                    jsonPaths[0].getPath(),
+                                    result0.size(),
+                                    jsonPaths[i].getPath(),
+                                    result.size()));
                 }
             }
         }
-
         return dataFlip(results);
     }