[Hotfix][File-Connector] Fix WriteStrategy parallel writing thread unsafe issue (#5546)

Thread unsafe exception occurs when parallel writes are run on a single node,
should not shares thread unsafe object to writers.

Author: hailin0

release-note.md

@@ -49,6 +49,7 @@
 - [Connector-v2] [Jdbc] Fix oracle sql table identifier (#4754)
 - [Connector-v2] [Clickhouse] fix get clickhouse local table name with closing bracket from distributed table engineFull (#4710)
 - [Connector-v2] [CDC] Fix jdbc connection leak for mysql (#5037)
+- [Connector-v2] [File] Fix WriteStrategy parallel writing thread unsafe issue #5546
 
 ### Zeta(ST-Engine)
 

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSink.java

@@ -51,7 +51,6 @@ public abstract class BaseFileSink
     protected HadoopConf hadoopConf;
     protected FileSystemUtils fileSystemUtils;
     protected FileSinkConfig fileSinkConfig;
-    protected WriteStrategy writeStrategy;
     protected JobContext jobContext;
     protected String jobId;
 
@@ -65,11 +64,7 @@ public void setJobContext(JobContext jobContext) {
     public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
         this.seaTunnelRowType = seaTunnelRowType;
         this.fileSinkConfig = new FileSinkConfig(pluginConfig, seaTunnelRowType);
-        this.writeStrategy =
-                WriteStrategyFactory.of(fileSinkConfig.getFileFormat(), fileSinkConfig);
         this.fileSystemUtils = new FileSystemUtils(hadoopConf);
-        this.writeStrategy.setSeaTunnelRowTypeInfo(seaTunnelRowType);
-        this.writeStrategy.setFileSystemUtils(fileSystemUtils);
     }
 
     @Override
@@ -80,7 +75,7 @@ public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
     @Override
     public SinkWriter<SeaTunnelRow, FileCommitInfo, FileSinkState> restoreWriter(
             SinkWriter.Context context, List<FileSinkState> states) throws IOException {
-        return new BaseFileSinkWriter(writeStrategy, hadoopConf, context, jobId, states);
+        return new BaseFileSinkWriter(createWriteStrategy(), hadoopConf, context, jobId, states);
     }
 
     @Override
@@ -92,7 +87,7 @@ public SinkWriter<SeaTunnelRow, FileCommitInfo, FileSinkState> restoreWriter(
     @Override
     public SinkWriter<SeaTunnelRow, FileCommitInfo, FileSinkState> createWriter(
             SinkWriter.Context context) throws IOException {
-        return new BaseFileSinkWriter(writeStrategy, hadoopConf, context, jobId);
+        return new BaseFileSinkWriter(createWriteStrategy(), hadoopConf, context, jobId);
     }
 
     @Override
@@ -121,4 +116,12 @@ public Optional<Serializer<FileSinkState>> getWriterStateSerializer() {
     public void prepare(Config pluginConfig) throws PrepareFailException {
         this.pluginConfig = pluginConfig;
     }
+
+    protected WriteStrategy createWriteStrategy() {
+        WriteStrategy writeStrategy =
+                WriteStrategyFactory.of(fileSinkConfig.getFileFormat(), fileSinkConfig);
+        writeStrategy.setSeaTunnelRowTypeInfo(seaTunnelRowType);
+        writeStrategy.setFileSystemUtils(fileSystemUtils);
+        return writeStrategy;
+    }
 }