[improve] Supports http request use utf8 charset

Author: gnehil

spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/AbstractStreamLoadProcessor.java

@@ -326,12 +326,8 @@ private void handleStreamLoadProperties(HttpPut httpPut) throws OptionRequiredEx
             currentLabel = generateStreamLoadLabel();
             httpPut.setHeader("label", currentLabel);
         }
-        if (config.getValue(DorisOptions.DORIS_SINK_ADD_COLUMNS_HEADER)) {
-            String writeFields = getWriteFields();
-            httpPut.setHeader("columns", writeFields);
-        } else {
-            logger.info("skip setting columns header for stream load");
-        }
+        String writeFields = getWriteFields();
+        httpPut.setHeader("columns", writeFields);
         if (config.contains(DorisOptions.DORIS_MAX_FILTER_RATIO)) {
             httpPut.setHeader("max_filter_ratio", config.getValue(DorisOptions.DORIS_MAX_FILTER_RATIO));
         }

spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisConfig.java

@@ -122,14 +122,6 @@ private void checkOptions(Map<String, String> options) throws OptionRequiredExce
                 throw new IllegalArgumentException("option [%s] and [%s] cannot be true at this same time");
             }
         }
-        if (options.containsKey(DorisOptions.DORIS_SINK_ADD_COLUMNS_HEADER.getName())
-                && !Boolean.parseBoolean(options.get(DorisOptions.DORIS_SINK_ADD_COLUMNS_HEADER.getName()))) {
-            Map<String, String> sinkProperties = getSinkProperties();
-            if (!sinkProperties.containsKey("format") || !"json".equalsIgnoreCase(sinkProperties.get("format"))) {
-                throw new IllegalArgumentException(String.format("option [%s] is only supported when sink property 'format' is 'json'",
-                        DorisOptions.DORIS_SINK_ADD_COLUMNS_HEADER.getName()));
-            }
-        }
     }
 
     public boolean contains(ConfigOption<?> option) {

spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/config/DorisOptions.java

@@ -144,7 +144,7 @@ public class DorisOptions {
 
     public static final ConfigOption<Integer> DORIS_SINK_NET_BUFFER_SIZE = ConfigOptions.name("doris.sink.net.buffer.size").intType().defaultValue(1024 * 1024).withDescription("");
 
-    public static final ConfigOption<Boolean> DORIS_SINK_ADD_COLUMNS_HEADER = ConfigOptions.name("doris.sink.add-columns-header").booleanType().defaultValue(true).withDescription("");
+    public static final ConfigOption<Boolean> DORIS_SINK_HTTP_UTF8_CHARSET = ConfigOptions.name("doris.sink.http-utf8-charset").booleanType().defaultValue(false).withDescription("");
 
 
 }

spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/util/HttpUtils.scala

@@ -20,21 +20,31 @@ package org.apache.doris.spark.util
 import org.apache.doris.spark.config.{DorisConfig, DorisOptions}
 import org.apache.http.HttpHeaders
 import org.apache.http.client.methods.HttpRequestBase
+import org.apache.http.config.ConnectionConfig
 import org.apache.http.conn.ssl.{SSLConnectionSocketFactory, TrustAllStrategy}
 import org.apache.http.impl.client.{CloseableHttpClient, DefaultRedirectStrategy, HttpClients}
 import org.apache.http.protocol.HttpRequestExecutor
 import org.apache.http.ssl.SSLContexts
 
 import java.io.{File, FileInputStream}
-import java.nio.charset.StandardCharsets
+import java.nio.charset.{CodingErrorAction, StandardCharsets}
 import java.security.KeyStore
 import java.util.Base64
 import scala.util.{Failure, Success, Try}
 
 object HttpUtils {
 
   def getHttpClient(config: DorisConfig): CloseableHttpClient = {
+
+    var connectionConfig = ConnectionConfig.DEFAULT;
+    if (config.getValue(DorisOptions.DORIS_SINK_HTTP_UTF8_CHARSET)) {
+      connectionConfig = ConnectionConfig.custom()
+        .setCharset(StandardCharsets.UTF_8)
+        .setMalformedInputAction(CodingErrorAction.REPLACE)
+        .setUnmappableInputAction(CodingErrorAction.REPLACE).build()
+    }
     val builder = HttpClients.custom()
+      .setDefaultConnectionConfig(connectionConfig)
       .setRequestExecutor(new HttpRequestExecutor(60000))
       .setRedirectStrategy(new DefaultRedirectStrategy {
         override def isRedirectable(method: String): Boolean = true

spark-doris-connector/spark-doris-connector-base/src/test/scala/org/apache/doris/spark/config/DorisConfigTest.scala

@@ -440,8 +440,7 @@ class DorisWriterITCase extends AbstractContainerTestBase {
            | "fenodes"="${getFenodes}",
            | "user"="${getDorisUsername}",
            | "password"="${getDorisPassword}",
-           | "doris.sink.add-columns-header"="false",
-           | "doris.sink.properties.format"="json"
+           | "doris.sink.http-utf8-charset"="true"
            |)
            |""".stripMargin)
       session.sql(