Remove same code to independent method in HiveSinkWriter (#2307)

Author: zhaomin1423

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/sink/HiveSink.java

@@ -113,7 +113,7 @@ public Optional<Serializer<HiveSinkState>> getWriterStateSerializer() {
 
     @Override
     public Optional<Serializer<HiveAggregatedCommitInfo>> getAggregatedCommitInfoSerializer() {
-        return Optional.of(new DefaultSerializer<HiveAggregatedCommitInfo>());
+        return Optional.of(new DefaultSerializer<>());
     }
 
     @Override

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/sink/HiveSinkWriter.java

@@ -63,31 +63,7 @@ public HiveSinkWriter(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
         this.context = context;
         this.jobId = jobId;
         this.hiveSinkConfig = hiveSinkConfig;
-
-        SinkFileSystemPlugin sinkFileSystemPlugin = new HdfsFileSinkPlugin();
-        Optional<TransactionStateFileWriter> transactionStateFileWriter = sinkFileSystemPlugin.getTransactionStateFileWriter(this.seaTunnelRowTypeInfo,
-            new FileSinkTransactionFileNameGenerator(
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileFormat(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameExpression(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameTimeFormat()),
-            new FileSinkPartitionDirNameGenerator(
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldList(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldsIndexInRow(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionDirExpression()),
-            this.hiveSinkConfig.getTextFileSinkConfig().getSinkColumnsIndexInRow(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getTmpPath(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getPath(),
-            this.jobId,
-            this.context.getIndexOfSubtask(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getFieldDelimiter(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getRowDelimiter(),
-            sinkFileSystemPlugin.getFileSystem().get());
-
-        if (!transactionStateFileWriter.isPresent()) {
-            throw new RuntimeException("A TransactionStateFileWriter is need");
-        }
-
-        this.fileWriter = transactionStateFileWriter.get();
+        this.fileWriter = createFileWriter();
 
         fileWriter.beginTransaction(1L);
     }
@@ -103,31 +79,7 @@ public HiveSinkWriter(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
         this.context = context;
         this.jobId = jobId;
         this.hiveSinkConfig = hiveSinkConfig;
-
-        SinkFileSystemPlugin sinkFileSystemPlugin = new HdfsFileSinkPlugin();
-        Optional<TransactionStateFileWriter> transactionStateFileWriter = sinkFileSystemPlugin.getTransactionStateFileWriter(this.seaTunnelRowTypeInfo,
-            new FileSinkTransactionFileNameGenerator(
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileFormat(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameExpression(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameTimeFormat()),
-            new FileSinkPartitionDirNameGenerator(
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldList(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldsIndexInRow(),
-                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionDirExpression()),
-            this.hiveSinkConfig.getTextFileSinkConfig().getSinkColumnsIndexInRow(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getTmpPath(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getPath(),
-            this.jobId,
-            this.context.getIndexOfSubtask(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getFieldDelimiter(),
-            this.hiveSinkConfig.getTextFileSinkConfig().getRowDelimiter(),
-            sinkFileSystemPlugin.getFileSystem().get());
-
-        if (!transactionStateFileWriter.isPresent()) {
-            throw new RuntimeException("A TransactionStateFileWriter is need");
-        }
-
-        this.fileWriter = transactionStateFileWriter.get();
+        this.fileWriter = createFileWriter();
 
         // Rollback dirty transaction
         if (hiveSinkStates.size() > 0) {
@@ -172,4 +124,37 @@ public List<HiveSinkState> snapshotState(long checkpointId) throws IOException {
     public void abortPrepare() {
         fileWriter.abortTransaction();
     }
+
+    private TransactionStateFileWriter createFileWriter() {
+        SinkFileSystemPlugin sinkFileSystemPlugin = new HdfsFileSinkPlugin();
+        Optional<TransactionStateFileWriter> transactionStateFileWriterOpt = sinkFileSystemPlugin.getTransactionStateFileWriter(this.seaTunnelRowTypeInfo,
+                getFilenameGenerator(),
+                getPartitionDirNameGenerator(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getSinkColumnsIndexInRow(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getTmpPath(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getPath(),
+                this.jobId,
+                this.context.getIndexOfSubtask(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getFieldDelimiter(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getRowDelimiter(),
+                sinkFileSystemPlugin.getFileSystem().get());
+        if (!transactionStateFileWriterOpt.isPresent()) {
+            throw new RuntimeException("A TransactionStateFileWriter is need");
+        }
+        return transactionStateFileWriterOpt.get();
+    }
+
+    private FileSinkTransactionFileNameGenerator getFilenameGenerator() {
+        return new FileSinkTransactionFileNameGenerator(
+                this.hiveSinkConfig.getTextFileSinkConfig().getFileFormat(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameExpression(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getFileNameTimeFormat());
+    }
+
+    private FileSinkPartitionDirNameGenerator getPartitionDirNameGenerator() {
+        return new FileSinkPartitionDirNameGenerator(
+                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldList(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionFieldsIndexInRow(),
+                this.hiveSinkConfig.getTextFileSinkConfig().getPartitionDirExpression());
+    }
 }