This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ba53e82bab [Fix][Connector-V2] Ignore BOM in file source readers
(#11056)
ba53e82bab is described below
commit ba53e82babde02fc2d8c411ae0a94d71a3e9648f
Author: shown <[email protected]>
AuthorDate: Sat Jun 13 13:44:16 2026 +0800
[Fix][Connector-V2] Ignore BOM in file source readers (#11056)
Signed-off-by: yuluo-yx <[email protected]>
---
.../file/source/reader/AbstractReadStrategy.java | 21 ++++++
.../file/source/reader/CsvReadStrategy.java | 15 +---
.../file/source/reader/JsonReadStrategy.java | 6 +-
.../file/source/reader/TextReadStrategy.java | 6 +-
.../file/source/reader/XmlReadStrategy.java | 6 +-
.../source/reader/AbstractReadStrategyTest.java | 80 ++++++++++++++++++++++
6 files changed, 111 insertions(+), 23 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
index a685a69fec..918e4641b5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
@@ -47,6 +47,8 @@ import
org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.apache.commons.compress.compressors.gzip.GzipParameters;
+import org.apache.commons.io.ByteOrderMark;
+import org.apache.commons.io.input.BOMInputStream;
import org.apache.commons.io.input.BoundedInputStream;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileStatus;
@@ -55,12 +57,15 @@ import org.apache.hadoop.fs.Seekable;
import lombok.extern.slf4j.Slf4j;
+import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InputStreamReader;
import java.math.BigDecimal;
+import java.nio.charset.Charset;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
@@ -602,6 +607,22 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
return new BoundedInputStream(in, length);
}
+ protected static BufferedReader createBomAwareBufferedReader(
+ InputStream inputStream, String encoding) throws IOException {
+ BOMInputStream bomInputStream =
+ new BOMInputStream(
+ inputStream,
+ ByteOrderMark.UTF_8,
+ ByteOrderMark.UTF_16BE,
+ ByteOrderMark.UTF_16LE,
+ ByteOrderMark.UTF_32BE,
+ ByteOrderMark.UTF_32LE);
+ ByteOrderMark bom = bomInputStream.getBOM();
+ Charset charset =
+ bom == null ? Charset.forName(encoding) :
Charset.forName(bom.getCharsetName());
+ return new BufferedReader(new InputStreamReader(bomInputStream,
charset));
+ }
+
@Override
public void close() throws IOException {
try {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/CsvReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/CsvReadStrategy.java
index a049d9fa55..7ac420a7cc 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/CsvReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/CsvReadStrategy.java
@@ -45,7 +45,6 @@ import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVFormat.Builder;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
-import org.apache.commons.io.input.BOMInputStream;
import io.airlift.compress.lzo.LzopCodec;
import lombok.extern.slf4j.Slf4j;
@@ -53,8 +52,6 @@ import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
@@ -107,9 +104,9 @@ public class CsvReadStrategy extends AbstractReadStrategy {
split.getStart(),
split.getLength());
final boolean useSplitRead = isSplitReadEnabled(split);
- try (BOMInputStream bomIn = new
BOMInputStream(wrapInputStream(inputStream, split));
- BufferedReader reader =
- new BufferedReader(new InputStreamReader(bomIn,
getCharset(bomIn)));
+ try (BufferedReader reader =
+ createBomAwareBufferedReader(
+ wrapInputStream(inputStream, split), encoding);
CSVParser csvParser = new CSVParser(reader,
getCSVFormat(split))) {
// skip lines
// if split range is used, no need to skip
@@ -204,12 +201,6 @@ public class CsvReadStrategy extends AbstractReadStrategy {
return resultStream;
}
- private Charset getCharset(BOMInputStream bomIn) throws IOException {
- return bomIn.getBOM() == null
- ? Charset.forName(encoding)
- : Charset.forName(bomIn.getBOM().getCharsetName());
- }
-
private boolean isSplitReadEnabled(FileSourceSplit split) {
return enableSplitFile && split.getLength() > -1;
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/JsonReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/JsonReadStrategy.java
index 98b8de3ca2..a3bca9c5b1 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/JsonReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/JsonReadStrategy.java
@@ -39,7 +39,6 @@ import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
@@ -116,10 +115,9 @@ public class JsonReadStrategy extends AbstractReadStrategy
{
}
// rebuild inputStream
if (enableSplitFile && split.getLength() > -1) {
- actualInputStream = safeSlice(inputStream, split.getStart(),
split.getLength());
+ actualInputStream = safeSlice(actualInputStream, split.getStart(),
split.getLength());
}
- try (BufferedReader reader =
- new BufferedReader(new InputStreamReader(actualInputStream,
encoding))) {
+ try (BufferedReader reader =
createBomAwareBufferedReader(actualInputStream, encoding)) {
reader.lines()
.forEach(
line -> {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/TextReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/TextReadStrategy.java
index fa45521fbb..4c6f078362 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/TextReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/TextReadStrategy.java
@@ -46,7 +46,6 @@ import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Optional;
@@ -207,10 +206,9 @@ public class TextReadStrategy extends AbstractReadStrategy
{
// rebuild inputStream
final boolean useSplitRead = enableSplitFile && split.getLength() > -1;
if (useSplitRead) {
- actualInputStream = safeSlice(inputStream, split.getStart(),
split.getLength());
+ actualInputStream = safeSlice(actualInputStream, split.getStart(),
split.getLength());
}
- try (BufferedReader reader =
- new BufferedReader(new InputStreamReader(actualInputStream,
encoding))) {
+ try (BufferedReader reader =
createBomAwareBufferedReader(actualInputStream, encoding)) {
LineProcessor lineProcessor =
line -> {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/XmlReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/XmlReadStrategy.java
index 58002f0f54..fcf7daf2dd 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/XmlReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/XmlReadStrategy.java
@@ -52,9 +52,9 @@ import org.dom4j.io.SAXReader;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -106,8 +106,8 @@ public class XmlReadStrategy extends AbstractReadStrategy {
throws IOException {
SAXReader saxReader = new SAXReader();
Document document;
- try {
- document = saxReader.read(new InputStreamReader(inputStream,
encoding));
+ try (BufferedReader reader = createBomAwareBufferedReader(inputStream,
encoding)) {
+ document = saxReader.read(reader);
} catch (DocumentException e) {
throw new FileConnectorException(
FileConnectorErrorCode.FILE_READ_FAILED,
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
index 626a33bbd7..69481f30d5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
@@ -26,6 +26,7 @@ import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
import org.apache.seatunnel.connectors.seatunnel.file.util.LocalFileSystemConf;
import org.apache.avro.Schema;
@@ -46,6 +47,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
+import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
@@ -381,6 +383,84 @@ public class AbstractReadStrategyTest {
}
}
+ @Test
+ void testTextReadStrategyShouldSkipUtf8Bom() throws Exception {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"name"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("test",
rowType);
+ TempCollector collector = new TempCollector();
+
+ try (TextReadStrategy textReadStrategy = new TextReadStrategy()) {
+ textReadStrategy.setPluginConfig(ConfigFactory.empty());
+ textReadStrategy.setCatalogTable(catalogTable);
+ textReadStrategy.readProcess(
+ new FileSourceSplit("test", "/tmp/bom.txt"),
+ collector,
+ new ByteArrayInputStream(
+ ("\uFEFF" +
"alice\n").getBytes(StandardCharsets.UTF_8)),
+ new HashMap<>(),
+ "bom.txt");
+ }
+
+ Assertions.assertEquals(1, collector.getRows().size());
+ Assertions.assertEquals("alice",
collector.getRows().get(0).getField(0));
+ }
+
+ @Test
+ void testJsonReadStrategyShouldSkipUtf8Bom() throws Exception {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"name"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("test",
rowType);
+ TempCollector collector = new TempCollector();
+
+ try (JsonReadStrategy jsonReadStrategy = new JsonReadStrategy()) {
+ jsonReadStrategy.setPluginConfig(ConfigFactory.empty());
+ jsonReadStrategy.init(new
LocalFileSystemConf.LocalConf(FS_DEFAULT_NAME_DEFAULT));
+ jsonReadStrategy.setCatalogTable(catalogTable);
+ jsonReadStrategy.readProcess(
+ new FileSourceSplit("test", "/tmp/bom.json"),
+ collector,
+ new ByteArrayInputStream(
+ ("\uFEFF" +
"{\"name\":\"alice\"}\n").getBytes(StandardCharsets.UTF_8)),
+ new HashMap<>(),
+ "bom.json");
+ }
+
+ Assertions.assertEquals(1, collector.getRows().size());
+ Assertions.assertEquals("alice",
collector.getRows().get(0).getField(0));
+ }
+
+ @Test
+ void testXmlReadStrategyShouldSkipUtf8Bom() throws Exception {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"name"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("test",
rowType);
+ Map<String, Object> pluginConfig = new HashMap<>();
+ pluginConfig.put(FileBaseSourceOptions.XML_ROW_TAG.key(), "row");
+ pluginConfig.put(FileBaseSourceOptions.XML_USE_ATTR_FORMAT.key(),
false);
+ TempCollector collector = new TempCollector();
+
+ try (XmlReadStrategy xmlReadStrategy = new XmlReadStrategy()) {
+
xmlReadStrategy.setPluginConfig(ConfigFactory.parseMap(pluginConfig));
+ xmlReadStrategy.init(new
LocalFileSystemConf.LocalConf(FS_DEFAULT_NAME_DEFAULT));
+ xmlReadStrategy.setCatalogTable(catalogTable);
+ xmlReadStrategy.readProcess(
+ new FileSourceSplit("test", "/tmp/bom.xml"),
+ collector,
+ new ByteArrayInputStream(
+ ("\uFEFF" +
"<rows><row><name>alice</name></row></rows>")
+ .getBytes(StandardCharsets.UTF_8)),
+ new HashMap<>(),
+ "bom.xml");
+ }
+
+ Assertions.assertEquals(1, collector.getRows().size());
+ Assertions.assertEquals("alice",
collector.getRows().get(0).getField(0));
+ }
+
@Test
void testResolveRelativePathWithSftpUri() {
String basePath = "sftp://server:22/path";;
