[fix]avoid label conflict when setting labels in headers. (#25)

Author: Petrichor

build.sh

@@ -19,6 +19,8 @@
 ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
 
 rm -rf version.go
+# Formatting Go code with gofmt
+find . -name '*.go' -exec gofmt -w {} \;
 go generate
 go build
 echo "Build success. Output: ${ROOT}/doris-streamloader"

loader/stream_loader.go

@@ -18,11 +18,13 @@
 package loader
 
 import (
+	"doris-streamloader/report"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -31,7 +33,6 @@ import (
 
 	"github.com/pierrec/lz4/v4"
 	log "github.com/sirupsen/logrus"
-	"doris-streamloader/report"
 )
 
 type StreamLoadOption struct {
@@ -143,7 +144,7 @@ func (s *StreamLoad) createUrl() string {
 }
 
 // stream load create http request with string data
-func (s *StreamLoad) createRequest(url string, reader io.Reader) (req *http.Request, err error) {
+func (s *StreamLoad) createRequest(url string, reader io.Reader, workerIndex int, taskIndex int) (req *http.Request, err error) {
 	req, err = http.NewRequest("PUT", url, reader)
 	if err != nil {
 		return
@@ -155,6 +156,19 @@ func (s *StreamLoad) createRequest(url string, reader io.Reader) (req *http.Requ
 	req.Header.Set("Content-Type", "text/plain")
 	for k, v := range s.headers {
 		req.Header.Set(k, v)
+		// If a label has already been set in the headers, to prevent conflicts,
+		//generate a unique label by combining the original label, worker index, and task index.
+		if k == "label" {
+			var builder strings.Builder
+			builder.WriteString(v)
+			builder.WriteString("_")
+			builder.WriteString(strconv.Itoa(workerIndex))
+			builder.WriteString("_")
+			builder.WriteString(strconv.Itoa(taskIndex))
+
+			req.Header.Set("label", builder.String())
+		}
+
 	}
 
 	if s.Compress {
@@ -228,10 +242,10 @@ func (s *StreamLoad) readData(isEOS *atomic.Bool, rawWriter *io.PipeWriter, read
 	}
 }
 
-func (s *StreamLoad) send(url string, reader io.Reader) (*http.Response, error) {
+func (s *StreamLoad) send(url string, reader io.Reader, workerIndex int, taskIndex int) (*http.Response, error) {
 	realUrl := url
 	for {
-		req, err := s.createRequest(realUrl, reader)
+		req, err := s.createRequest(realUrl, reader, workerIndex, taskIndex)
 		if err != nil {
 			if req == nil {
 				return nil, err
@@ -347,7 +361,7 @@ func (s *StreamLoad) executeGetAndSend(maxRowsPerTask int, maxBytesPerTask int,
 				workerIndex:     workerIndex,
 				taskIndex:       taskIndex,
 			})
-		if resp, err := s.send(url, NopCloser(pr)); err != nil {
+		if resp, err := s.send(url, NopCloser(pr), workerIndex, taskIndex); err != nil {
 			s.handleSendError(workerIndex, taskIndex)
 			log.Errorf("Send error, resp: %v error message: %v", resp, err)
 			return

reader/reader.go

@@ -19,6 +19,8 @@ package file
 
 import (
 	"bufio"
+	"doris-streamloader/loader"
+	"doris-streamloader/report"
 	"io"
 	"os"
 	"path/filepath"
@@ -28,8 +30,6 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
-	report "doris-streamloader/report"
-	loader "doris-streamloader/loader"
 )
 
 type FileReader struct {