[improve] Supports http request use utf8 charset

gnehil · gnehil · commit d60e73e6136c · 2025-11-21T15:26:32.000+08:00
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
@@ -60,4 +60,8 @@ public interface ConfigurationOptions {
 
     String FLIGHT_SQL_PORT = "source.flight-sql-port";
     Integer FLIGHT_SQL_PORT_DEFAULT = -1;
+
+    String SINK_HTTP_UTF8_CHARSET = "sink.http-utf8-charset";
+    Boolean SINK_HTTP_UTF8_CHARSET_DEFAULT = false;
+
 }
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java
@@ -49,6 +49,7 @@ public class DorisExecutionOptions implements Serializable {
     private final int bufferCount;
     private final String labelPrefix;
     private final boolean useCache;
+    private final boolean httpUtf8Charset;
 
     /** Properties for the StreamLoad. */
     private final Properties streamLoadProp;
@@ -74,6 +75,7 @@ public DorisExecutionOptions(
             int bufferCount,
             String labelPrefix,
             boolean useCache,
+            boolean httpUtf8Charset,
             Properties streamLoadProp,
             Boolean enableDelete,
             Boolean enable2PC,
@@ -93,6 +95,7 @@ public DorisExecutionOptions(
         this.bufferCount = bufferCount;
         this.labelPrefix = labelPrefix;
         this.useCache = useCache;
+        this.httpUtf8Charset = httpUtf8Charset;
         this.streamLoadProp = streamLoadProp;
         this.enableDelete = enableDelete;
         this.enable2PC = enable2PC;
@@ -162,6 +165,10 @@ public boolean isUseCache() {
         return useCache;
     }
 
+    public boolean isHttpUtf8Charset() {
+        return httpUtf8Charset;
+    }
+
     public Properties getStreamLoadProp() {
         return streamLoadProp;
     }
@@ -228,6 +235,7 @@ public boolean equals(Object o) {
                 && bufferSize == that.bufferSize
                 && bufferCount == that.bufferCount
                 && useCache == that.useCache
+                && httpUtf8Charset == that.httpUtf8Charset
                 && force2PC == that.force2PC
                 && flushQueueSize == that.flushQueueSize
                 && bufferFlushMaxRows == that.bufferFlushMaxRows
@@ -252,6 +260,7 @@ public int hashCode() {
                 bufferCount,
                 labelPrefix,
                 useCache,
+                httpUtf8Charset,
                 streamLoadProp,
                 enableDelete,
                 enable2PC,
@@ -274,6 +283,7 @@ public static class Builder {
         private int bufferCount = DEFAULT_BUFFER_COUNT;
         private String labelPrefix = "";
         private boolean useCache = false;
+        private boolean httpUtf8Charset = false;
         private Properties streamLoadProp = new Properties();
         private boolean enableDelete = true;
         private boolean enable2PC = true;
@@ -361,6 +371,17 @@ public Builder setUseCache(boolean useCache) {
             return this;
         }
 
+        /**
+         * Sets whether to set http utf8 charset for stream load.
+         *
+         * @param httpUtf8Charset
+         * @return this DorisExecutionOptions.builder.
+         */
+        public Builder setHttpUtf8Charset(boolean httpUtf8Charset) {
+            this.httpUtf8Charset = httpUtf8Charset;
+            return this;
+        }
+
         /**
          * Sets the properties for stream load.
          *
@@ -529,6 +550,7 @@ public DorisExecutionOptions build() {
                     bufferCount,
                     labelPrefix,
                     useCache,
+                    httpUtf8Charset,
                     streamLoadProp,
                     enableDelete,
                     enable2PC,
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/HttpUtil.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/HttpUtil.java
@@ -19,6 +19,7 @@
 
 import org.apache.doris.flink.cfg.DorisReadOptions;
 import org.apache.http.client.config.RequestConfig;
+import org.apache.http.config.ConnectionConfig;
 import org.apache.http.impl.NoConnectionReuseStrategy;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.DefaultRedirectStrategy;
@@ -27,29 +28,45 @@
 import org.apache.http.protocol.HttpRequestExecutor;
 import org.apache.http.protocol.RequestContent;
 
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_REQUEST_CONNECT_TIMEOUT_MS_DEFAULT;
+import static org.apache.doris.flink.cfg.ConfigurationOptions.SINK_HTTP_UTF8_CHARSET_DEFAULT;
 
 /** util to build http client. */
 public class HttpUtil {
     private final int connectTimeout;
     private final int waitForContinueTimeout;
+    private final boolean httpUtf8Charset;
     private HttpClientBuilder httpClientBuilder;
 
     public HttpUtil() {
         this.connectTimeout = DORIS_REQUEST_CONNECT_TIMEOUT_MS_DEFAULT;
         this.waitForContinueTimeout = DORIS_REQUEST_CONNECT_TIMEOUT_MS_DEFAULT;
+        this.httpUtf8Charset = SINK_HTTP_UTF8_CHARSET_DEFAULT;
         settingStreamHttpClientBuilder();
     }
 
-    public HttpUtil(DorisReadOptions readOptions) {
+    public HttpUtil(DorisReadOptions readOptions, boolean httpUtf8Charset) {
         this.connectTimeout = readOptions.getRequestConnectTimeoutMs();
         this.waitForContinueTimeout = readOptions.getRequestConnectTimeoutMs();
+        this.httpUtf8Charset = httpUtf8Charset;
         settingStreamHttpClientBuilder();
     }
 
     private void settingStreamHttpClientBuilder() {
+        ConnectionConfig connectionConfig = ConnectionConfig.DEFAULT;
+        if (httpUtf8Charset) {
+            connectionConfig = ConnectionConfig.custom()
+                    .setCharset(StandardCharsets.UTF_8)
+                    .setMalformedInputAction(CodingErrorAction.REPLACE)
+                    .setUnmappableInputAction(CodingErrorAction.REPLACE)
+                    .build();
+        }
         this.httpClientBuilder =
                 HttpClients.custom()
+                        .setDefaultConnectionConfig(connectionConfig)
                         // default timeout 3s, maybe report 307 error when fe busy
                         .setRequestExecutor(new HttpRequestExecutor(waitForContinueTimeout))
                         .setRedirectStrategy(
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
@@ -172,7 +172,7 @@ public DorisBatchStreamLoad(
         this.started = new AtomicBoolean(true);
         this.loadExecutorService.execute(loadAsyncExecutor);
         this.subTaskId = subTaskId;
-        this.httpClientBuilder = new HttpUtil(dorisReadOptions).getHttpClientBuilderForBatch();
+        this.httpClientBuilder = new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClientBuilderForBatch();
     }
 
     /**
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
@@ -70,7 +70,7 @@ public DorisCommitter(
                 dorisOptions,
                 dorisReadOptions,
                 executionOptions,
-                new HttpUtil(dorisReadOptions).getHttpClient());
+                new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClient());
     }
 
     public DorisCommitter(
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
@@ -310,7 +310,7 @@ public DorisStreamLoad getStreamLoader(String tableKey) {
                                 dorisOptions,
                                 executionOptions,
                                 labelGenerator,
-                                new HttpUtil(dorisReadOptions).getHttpClient()));
+                                new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClient()));
     }
 
     /** Http throws an exception actively, there is no need to check regularly. */
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
@@ -40,7 +40,9 @@
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_TABLET_SIZE_DEFAULT;
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_THRIFT_MAX_MESSAGE_SIZE_DEFAULT;
 
-/** Options for the Doris connector. */
+/**
+ * Options for the Doris connector.
+ */
 @PublicEvolving
 public class DorisConfigOptions {
 
@@ -340,4 +342,11 @@ public static Properties getStreamLoadProp(Map<String, String> tableOptions) {
         }
         return streamLoadProp;
     }
+
+    public static final ConfigOption<Boolean> SINK_HTTP_UTF8_CHARSET =
+            ConfigOptions.key("sink.http-utf8-charset")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Set sink http client default charset to utf8 for support unicode characters in header, the default value is false");
+
 }
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
@@ -86,6 +86,7 @@
 import static org.apache.doris.flink.table.DorisConfigOptions.TABLE_IDENTIFIER;
 import static org.apache.doris.flink.table.DorisConfigOptions.USERNAME;
 import static org.apache.doris.flink.table.DorisConfigOptions.USE_FLIGHT_SQL;
+import static org.apache.doris.flink.table.DorisConfigOptions.SINK_HTTP_UTF8_CHARSET;
 
 /**
  * The {@link DorisDynamicTableFactory} translates the catalog table to a table source.
@@ -165,6 +166,7 @@ public Set<ConfigOption<?>> optionalOptions() {
 
         options.add(USE_FLIGHT_SQL);
         options.add(FLIGHT_SQL_PORT);
+        options.add(SINK_HTTP_UTF8_CHARSET);
         return options;
     }
 
@@ -263,6 +265,7 @@ private DorisExecutionOptions getDorisExecutionOptions(
                 (int) readableConfig.get(SINK_BUFFER_FLUSH_MAX_BYTES).getBytes());
         builder.setBufferFlushIntervalMs(readableConfig.get(SINK_BUFFER_FLUSH_INTERVAL).toMillis());
         builder.setUseCache(readableConfig.get(SINK_USE_CACHE));
+        builder.setHttpUtf8Charset(readableConfig.get(SINK_HTTP_UTF8_CHARSET));
         return builder.build();
     }
 
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -326,6 +326,9 @@ public DorisSink<String> buildDorisSink(String tableIdentifier) {
         sinkConfig
                 .getOptional(DorisConfigOptions.SINK_IGNORE_COMMIT_ERROR)
                 .ifPresent(executionBuilder::setIgnoreCommitError);
+        sinkConfig
+                .getOptional(DorisConfigOptions.SINK_HTTP_UTF8_CHARSET)
+                .ifPresent(executionBuilder::setHttpUtf8Charset);
 
         DorisExecutionOptions executionOptions = executionBuilder.build();
         builder.setDorisReadOptions(DorisReadOptions.builder().build())
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/cfg/DorisExecutionOptionsTest.java
@@ -62,6 +62,7 @@ public void testEquals() {
                         .setUseCache(true)
                         .setFlushQueueSize(2)
                         .setIgnoreUpdateBefore(true)
+                        .setHttpUtf8Charset(true)
                         .build();
 
         DorisExecutionOptions.Builder builder =
@@ -81,7 +82,8 @@ public void testEquals() {
                         .setBatchMode(false)
                         .setUseCache(true)
                         .setFlushQueueSize(2)
-                        .setIgnoreUpdateBefore(true);
+                        .setIgnoreUpdateBefore(true)
+                        .setHttpUtf8Charset(true);
 
         Assert.assertNotEquals(exceptOptions, null);
         Assert.assertEquals(exceptOptions, exceptOptions);
@@ -147,9 +149,9 @@ public void testEquals() {
         Assert.assertNotEquals(exceptOptions, builder.build());
         builder.setFlushQueueSize(2);
 
-        builder.setIgnoreUpdateBefore(false);
+        builder.setHttpUtf8Charset(false);
         Assert.assertNotEquals(exceptOptions, builder.build());
-        builder.setIgnoreUpdateBefore(true);
+        builder.setHttpUtf8Charset(true);
     }
 
     @Test(expected = IllegalArgumentException.class)
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/sink/DorisSinkITCase.java
@@ -78,6 +78,7 @@ public class DorisSinkITCase extends AbstractITCaseService {
     static final String TABLE_GZ_FORMAT = "tbl_gz_format";
     static final String TABLE_CSV_JM = "tbl_csv_jm";
     static final String TABLE_CSV_TM = "tbl_csv_tm";
+    static final String TABLE_UNICODE_COLUMN = "tbl_unicode_column";
 
     private final boolean batchMode;
 
@@ -732,4 +733,82 @@ private void initializeFailoverTable(String table, DataModel dataModel) {
                         max,
                         model));
     }
+
+    private void initializeUnicodeColumnTable(String table, DataModel dataModel) {
+        String max = DataModel.AGGREGATE.equals(dataModel) ? "MAX" : "";
+        String morProps =
+                !DataModel.UNIQUE_MOR.equals(dataModel)
+                        ? ""
+                        : ",\"enable_unique_key_merge_on_write\" = \"false\"";
+        String model =
+                dataModel.equals(DataModel.UNIQUE_MOR)
+                        ? DataModel.UNIQUE.toString()
+                        : dataModel.toString();
+        ContainerUtils.executeSQLStatement(
+                getDorisQueryConnection(),
+                LOG,
+                String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE),
+                String.format("DROP TABLE IF EXISTS %s.%s", DATABASE, table),
+                "SET enable_unicode_name_support = true;",
+                String.format(
+                        "CREATE TABLE %s.%s ( \n"
+                                + "`名称` varchar(256),\n"
+                                + "`年龄` int %s\n"
+                                + ") "
+                                + " %s KEY(`名称`) "
+                                + " DISTRIBUTED BY HASH(`名称`) BUCKETS 1\n"
+                                + "PROPERTIES ("
+                                + "\"replication_num\" = \"1\"\n"
+                                + morProps
+                                + ")",
+                        DATABASE,
+                        table,
+                        max,
+                        model));
+    }
+
+    @Test
+    public void testSinkUnicodeColumn() throws Exception {
+        initializeUnicodeColumnTable(TABLE_UNICODE_COLUMN, DataModel.UNIQUE);
+        Properties properties = new Properties();
+        properties.setProperty("read_json_by_line", "true");
+        properties.setProperty("format", "json");
+
+        // mock data
+        Map<String, Object> row1 = new HashMap<>();
+        row1.put("名称", "doris1");
+        row1.put("年龄", 1);
+        Map<String, Object> row2 = new HashMap<>();
+        row2.put("名称", "doris2");
+        row2.put("年龄", 2);
+
+        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
+        executionBuilder
+                .setLabelPrefix(UUID.randomUUID().toString())
+                .setBatchMode(batchMode)
+                .setStreamLoadProp(properties)
+                // uniq need to be false
+                .setDeletable(false)
+                .setHttpUtf8Charset(true);
+        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
+        dorisBuilder
+                .setFenodes(getFenodes())
+                .setTableIdentifier(DATABASE + "." + TABLE_UNICODE_COLUMN)
+                .setUsername(getDorisUsername())
+                .setPassword(getDorisPassword());
+
+        submitJob(
+                dorisBuilder.build(),
+                executionBuilder.build(),
+                new String[] {
+                        new ObjectMapper().writeValueAsString(row1),
+                        new ObjectMapper().writeValueAsString(row2)
+                });
+
+        Thread.sleep(10000);
+        List<String> expected = Arrays.asList("doris1,1", "doris2,2");
+        String query = String.format("select `名称`,`年龄` from %s.%s order by 1", DATABASE, TABLE_UNICODE_COLUMN);
+        ContainerUtils.checkResult(getDorisQueryConnection(), LOG, expected, query, 2);
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -60,4 +60,8 @@ public interface ConfigurationOptions {`
`60`	`60`
`61`	`61`	`String FLIGHT_SQL_PORT = "source.flight-sql-port";`
`62`	`62`	`Integer FLIGHT_SQL_PORT_DEFAULT = -1;`
	`63`	`+`
	`64`	`+ String SINK_HTTP_UTF8_CHARSET = "sink.http-utf8-charset";`
	`65`	`+ Boolean SINK_HTTP_UTF8_CHARSET_DEFAULT = false;`
	`66`	`+`
`63`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -172,7 +172,7 @@ public DorisBatchStreamLoad(`
`172`	`172`	`this.started = new AtomicBoolean(true);`
`173`	`173`	`this.loadExecutorService.execute(loadAsyncExecutor);`
`174`	`174`	`this.subTaskId = subTaskId;`
`175`		`- this.httpClientBuilder = new HttpUtil(dorisReadOptions).getHttpClientBuilderForBatch();`
	`175`	`+ this.httpClientBuilder = new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClientBuilderForBatch();`
`176`	`176`	`}`
`177`	`177`
`178`	`178`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ public DorisCommitter(`
`70`	`70`	`dorisOptions,`
`71`	`71`	`dorisReadOptions,`
`72`	`72`	`executionOptions,`
`73`		`- new HttpUtil(dorisReadOptions).getHttpClient());`
	`73`	`+ new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClient());`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`public DorisCommitter(`
Original file line number	Diff line number	Diff line change
`@@ -310,7 +310,7 @@ public DorisStreamLoad getStreamLoader(String tableKey) {`
`310`	`310`	`dorisOptions,`
`311`	`311`	`executionOptions,`
`312`	`312`	`labelGenerator,`
`313`		`- new HttpUtil(dorisReadOptions).getHttpClient()));`
	`313`	`+ new HttpUtil(dorisReadOptions, executionOptions.isHttpUtf8Charset()).getHttpClient()));`
`314`	`314`	`}`
`315`	`315`
`316`	`316`	`/** Http throws an exception actively, there is no need to check regularly. */`