refactor(exporter): init fileWriter in cmd layer

cmd/exporter.go

@@ -7,6 +7,7 @@ import (
 	"github.com/huantt/kafka-dump/pkg/log"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
+	"github.com/xitongsys/parquet-go-source/local"
 	"sync"
 	"time"
 )
@@ -57,7 +58,11 @@ func CreateExportCommand() (*cobra.Command, error) {
 							outputFilePath = fmt.Sprintf("%s.%d", filePath, time.Now().UnixMilli())
 						}
 						log.Infof("[Worker-%d] Exporting to: %s", workerID, outputFilePath)
-						parquetWriter, err := impl.NewParquetWriter(outputFilePath)
+						fileWriter, err := local.NewLocalFileWriter(filePath)
+						if err != nil {
+							panic(errors.Wrap(err, "[NewLocalFileWriter]"))
+						}
+						parquetWriter, err := impl.NewParquetWriter(fileWriter)
 						if err != nil {
 							panic(errors.Wrap(err, "Unable to init parquet file writer"))
 						}

impl/parquet_writer.go

@@ -5,7 +5,6 @@ import (
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/huantt/kafka-dump/pkg/log"
 	"github.com/pkg/errors"
-	"github.com/xitongsys/parquet-go-source/local"
 	"github.com/xitongsys/parquet-go/source"
 	"github.com/xitongsys/parquet-go/writer"
 )
@@ -15,17 +14,13 @@ type ParquetWriter struct {
 	fileWriter    source.ParquetFile
 }
 
-func NewParquetWriter(filePath string) (*ParquetWriter, error) {
-	fw, err := local.NewLocalFileWriter(filePath)
-	if err != nil {
-		return nil, errors.Wrap(err, "[NewLocalFileWriter]")
-	}
-	parquetWriter, err := writer.NewParquetWriter(fw, new(ParquetMessage), 4)
+func NewParquetWriter(fileWriter source.ParquetFile) (*ParquetWriter, error) {
+	parquetWriter, err := writer.NewParquetWriter(fileWriter, new(ParquetMessage), 4)
 	if err != nil {
 		return nil, errors.Wrap(err, "[NewParquetWriter]")
 	}
 	return &ParquetWriter{
-		fileWriter:    fw,
+		fileWriter:    fileWriter,
 		parquetWriter: parquetWriter,
 	}, nil
 }