From 73a2334dc8a0aa821402471eaa19b64d96cf8b27 Mon Sep 17 00:00:00 2001
From: Wu Shilin <wushilin.sg@gmail.com>
Date: Fri, 28 Aug 2026 16:24:25 +0800
Subject: [PATCH] [Improve](go) Align stream load retry policy with the Flink
 connector

The Go SDK only retried HTTP 429/408/5xx. Doris reports almost every load
failure as HTTP 200 + Status "Fail" -- including transient ones such as
"[E-235] too many versions", "[E-238] too many segments", "current running
txns on db ... larger than limit", memory limit exceeded and txn/publish
timeouts -- so none of those were ever retried, while the Flink connector
(DorisBatchStreamLoad.load) retries every failed load.

Bring the Go SDK on par with the Flink connector:

- classifyResponseError: every non-success outcome (non-2xx HTTP, or 2xx
  with a Status other than Success/Publish Timeout) is retriable. The only
  exception is "Label Already Exists" + ExistingJobStatus RUNNING, which
  stays ambiguous and is resolved by polling get_load_state.
- New Config.MaxRetries / LoaderConfig.max_retries, default 3 (same as
  sink.max-retries); a batch is sent at most MaxRetries+1 times, still
  bounded by DorisUploadTimeout. Negative disables retries.
- Every retry uses a fresh label <label>_<attempt>, mirroring the Flink
  connector's label + "_" + retry.
- streamLoadError.Error() includes ErrorURL and keeps both the wrapper
  message and the last Doris response message.

Ambiguous outcomes (response lost after the request was sent, label
RUNNING) are still resolved through get_load_state before any retry, so
data is never loaded twice -- Flink batch mode re-sends blindly there.

Tests: updated classification table; new fakeSender tests for MaxRetries,
fresh labels and -1; new mock_backend_test.go drives the real HTTP path
against a scripted mock Doris (stream load + get_load_state, dropped
connections, FE 307 redirect) for each scenario: Fail retried, txn limit
exhausts MaxRetries, non-2xx retried, connection refused retried without
polling, dropped response VISIBLE not re-sent, ABORTED/UNKNOWN retried,
PREPARE polled, poll timeout fails without re-send, Label Already Exists
FINISHED/RUNNING/other, redirect keeps auth, concurrent batches.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UzpE493uFnJ1UQWx5N55sh
---
 .../doris_go_stream_load/README.md            |  23 +-
 .../doris_go_stream_load/client.go            |  87 ++-
 .../doris_go_stream_load/client_test.go       | 287 +++++++-
 .../doris_go_stream_load/config.go            |  33 +-
 .../doris_go_stream_load/loader_config.go     |   3 +
 .../loader_config_test.go                     |   4 +
 .../doris_go_stream_load/mock_backend_test.go | 685 ++++++++++++++++++
 .../doris_go_stream_load/sender.go            |  77 +-
 8 files changed, 1095 insertions(+), 104 deletions(-)
 create mode 100644 stream-load-sdks/doris_go_stream_load/mock_backend_test.go

diff --git a/stream-load-sdks/doris_go_stream_load/README.md b/stream-load-sdks/doris_go_stream_load/README.md
index 065e376..3b9ff38 100644
--- a/stream-load-sdks/doris_go_stream_load/README.md
+++ b/stream-load-sdks/doris_go_stream_load/README.md
@@ -304,6 +304,7 @@ Retry and timing:
 |---|---|---|
 | `DorisUploadRequestTimeout` | `300s` | HTTP deadline for one upload or label-poll request; minimum `10s` |
 | `DorisUploadTimeout` | `300s` | Total retry decision budget after retriable upload outcomes |
+| `MaxRetries` | `3` | Max retries per batch after a failed stream load (`MaxRetries+1` attempts in total); `-1` disables retries. Same semantics as the Flink connector's `sink.max-retries` |
 | `StatusPollTimeout` | `300s` | Max time spent polling a label after an ambiguous outcome |
 | `CallbackTimeout` | `100ms` | Reserved callback timing budget in config |
 | `SlowCallbackWarn` | `10ms` | Slow callback warning threshold |
@@ -356,13 +357,25 @@ Delivery can fail after queue admission; check `DeliveryResult.Err` from the han
 
 | Failure | Meaning | Typical action |
 |---|---|---|
-| HTTP 4xx from Doris | Bad URL, auth, table, schema, label, or data format | Inspect `StatusCode` and Doris response message |
-| HTTP 5xx or retriable transport error | Doris/BE/network may be unavailable | The SDK retries within `DorisUploadTimeout`; check cluster health |
-| ambiguous transport error | Request may have reached Doris but response was lost | The SDK polls the load label to decide whether it became visible |
-| label state `UNKNOWN` | Doris does not know the label | The transaction was not registered; final result is failure |
-| status poll timeout | Doris did not reach a final visible/failed state in time | Increase `StatusPollTimeout` or inspect Doris load jobs |
+| HTTP 200 with `Status: Fail` | Doris rejected the load. Transient causes include `[E-235] too many versions`, `[E-238] too many segments`, `current running txns on db ... larger than limit`, memory limit exceeded, txn/publish timeouts; permanent causes include schema/data-quality errors | Retried up to `MaxRetries` times with a fresh label; inspect `Response.Message` / `ErrorURL` if it still fails |
+| non-2xx HTTP status | Bad URL, auth, table, proxy/FE/BE unavailable | Retried up to `MaxRetries` times; inspect `StatusCode` and Doris response message |
+| connection refused / dial error | Request never reached Doris | Retried up to `MaxRetries` times (no polling needed) |
+| ambiguous transport error, or `Label Already Exists` with `ExistingJobStatus: RUNNING` | Request may have reached Doris but the response was lost | The SDK polls the load label: `VISIBLE`/`COMMITTED` is success, `ABORTED`/`UNKNOWN` is retried with a fresh label |
+| status poll timeout | Doris did not reach a final visible/failed state in time | Final result is failure (never retried, to avoid loading twice); increase `StatusPollTimeout` or inspect Doris load jobs |
+| `upload did not conclude within doris upload timeout` | `DorisUploadTimeout` elapsed while retrying | Raise `DorisUploadTimeout` or fix the underlying error |
 | callback too slow | Callback exceeded `SlowCallbackWarn` and produced an info log | Keep callbacks small; hand work to another goroutine if needed |
 
+### Retry policy
+
+The retry policy mirrors the Apache Doris Flink connector (`DorisBatchStreamLoad.load`):
+
+- Every failed stream load is retried, whatever the HTTP status or Doris `Status`/`Message`. Doris reports nearly all load failures as HTTP 200 + `Status: Fail`, and many of them (`[E-235] too many versions`, txn limit, memory limit, timeouts) are transient, so the SDK does not try to guess which messages are permanent.
+- A batch is attempted at most `MaxRetries + 1` times (default 3 retries, like `sink.max-retries`), and never beyond `DorisUploadTimeout`.
+- Each retry uses a fresh label `<label>_<attempt>` (`..._1`, `..._2`, ...), exactly like the Flink connector's `label + "_" + retry`.
+- Backoff between retries starts at 1s and doubles up to 4s.
+- `Success` and `Publish Timeout` are success; `Label Already Exists` with `ExistingJobStatus: FINISHED` is success.
+- Unlike the Flink connector, an outcome that is *ambiguous* (timeout or connection drop after the request was sent, or `Label Already Exists` with `RUNNING`) is resolved by polling `get_load_state` first, so data is never loaded twice.
+
 ## Shutdown
 
 `Close()` stops intake, drains accepted work, and waits for in-flight deliveries. After `Close()`, new sends return `ErrClientClosed`; already accepted handles still complete.
diff --git a/stream-load-sdks/doris_go_stream_load/client.go b/stream-load-sdks/doris_go_stream_load/client.go
index 14dd6c0..132c5c6 100644
--- a/stream-load-sdks/doris_go_stream_load/client.go
+++ b/stream-load-sdks/doris_go_stream_load/client.go
@@ -473,6 +473,19 @@ func (c *Client) runWorker(id int) {
 	}
 }
 
+// deliverBatch sends one batch to Doris and drives the retry loop. The retry
+// policy mirrors the Flink connector's DorisBatchStreamLoad.load():
+//
+//   - every failed load is retried, at most Config.MaxRetries times
+//     (MaxRetries+1 attempts in total), with a linear-ish backoff;
+//   - each retry uses a fresh label, <label>_<attempt>, because the failed
+//     attempt may have registered the previous label in Doris;
+//   - retrying also stops once Config.DorisUploadTimeout has elapsed.
+//
+// Unlike the Flink connector, an outcome that is ambiguous (the request may
+// have been applied but the response was lost, or the label is still RUNNING)
+// is resolved by polling the label state first; only ABORTED/UNKNOWN labels
+// are retried so that data is never loaded twice.
 func (c *Client) deliverBatch(batch *deliveryBatch) {
 	var (
 		outcome sendOutcome
@@ -481,43 +494,55 @@ func (c *Client) deliverBatch(batch *deliveryBatch) {
 
 	started := time.Now()
 	attempts := 0
+	maxRetries := c.cfg.maxRetries()
+	baseLabel := batch.label
 	var retryDeadline time.Time
 
+	fail := func(err error) {
+		c.completeBatch(batch, DeliveryResult{
+			Err:        err,
+			Attempts:   attempts,
+			StatusCode: outcome.statusCode,
+			Response:   outcome.response,
+			StartedAt:  started,
+			FinishedAt: time.Now(),
+		})
+	}
+	uploadTimeoutErr := func() error {
+		return &streamLoadError{
+			StatusCode: outcome.statusCode,
+			Message:    fmt.Sprintf("upload did not conclude within doris upload timeout %s", c.cfg.DorisUploadTimeout),
+			Response:   outcome.response,
+		}
+	}
+
 	for {
 		if attempts > 0 && !retryDeadline.IsZero() && time.Now().After(retryDeadline) {
-			result := DeliveryResult{
-				Err:        &streamLoadError{StatusCode: outcome.statusCode, Message: fmt.Sprintf("upload did not conclude within doris upload timeout %s", c.cfg.DorisUploadTimeout)},
-				Attempts:   attempts,
-				StatusCode: outcome.statusCode,
-				Response:   outcome.response,
-				StartedAt:  started,
-				FinishedAt: time.Now(),
-			}
-			c.completeBatch(batch, result)
+			fail(uploadTimeoutErr())
 			return
 		}
 
 		attempts++
 		c.stats.recordUploadAttempt(batch.byteSize, batch.len())
+		c.logf(LogLevelDebug, "stream load started: label=%s attempt=%d", batch.label, attempts)
 		ctx, cancel := context.WithTimeout(context.Background(), c.cfg.DorisUploadRequestTimeout)
 		outcome, err = c.sender.Send(ctx, batch)
 		cancel()
 
 		if err == nil {
-			result := DeliveryResult{
+			c.completeBatch(batch, DeliveryResult{
 				Attempts:   attempts,
 				StatusCode: outcome.statusCode,
 				Response:   outcome.response,
 				StartedAt:  started,
 				FinishedAt: time.Now(),
-			}
-			c.completeBatch(batch, result)
+			})
 			return
 		}
 
 		retriable := isRetriable(err)
 		ambiguous := isAmbiguous(err)
-		c.logf(LogLevelDebug, "send error: attempts=%d retriable=%t err=%v", attempts, retriable, err)
+		c.logf(LogLevelDebug, "send error: label=%s attempts=%d retriable=%t ambiguous=%t err=%v", batch.label, attempts, retriable, ambiguous, err)
 		if ambiguous {
 			pollResult, pollErr := c.pollLabelUntilConclusion(batch.label, started, attempts)
 			if pollErr == nil {
@@ -526,22 +551,14 @@ func (c *Client) deliverBatch(batch *deliveryBatch) {
 			}
 			err = pollErr
 			retriable = isRetriable(pollErr)
-			if retriable {
-				// Generate a fresh label: the old one is now registered in Doris
-				// (as ABORTED) and cannot be reused.
-				batch.label = generateLabel(c.cfg.LabelPrefix)
-			}
 		}
 		if !retriable {
-			result := DeliveryResult{
-				Err:        err,
-				Attempts:   attempts,
-				StatusCode: outcome.statusCode,
-				Response:   outcome.response,
-				StartedAt:  started,
-				FinishedAt: time.Now(),
-			}
-			c.completeBatch(batch, result)
+			fail(err)
+			return
+		}
+		if attempts > maxRetries {
+			c.logf(LogLevelError, "stream load failed: label=%s attempts=%d max_retries=%d, giving up: %v", batch.label, attempts, maxRetries, err)
+			fail(err)
 			return
 		}
 
@@ -549,15 +566,7 @@ func (c *Client) deliverBatch(batch *deliveryBatch) {
 			retryDeadline = time.Now().Add(c.cfg.DorisUploadTimeout)
 		}
 		if time.Now().After(retryDeadline) {
-			result := DeliveryResult{
-				Err:        &streamLoadError{StatusCode: outcome.statusCode, Message: fmt.Sprintf("upload did not conclude within doris upload timeout %s", c.cfg.DorisUploadTimeout)},
-				Attempts:   attempts,
-				StatusCode: outcome.statusCode,
-				Response:   outcome.response,
-				StartedAt:  started,
-				FinishedAt: time.Now(),
-			}
-			c.completeBatch(batch, result)
+			fail(uploadTimeoutErr())
 			return
 		}
 
@@ -565,7 +574,11 @@ func (c *Client) deliverBatch(batch *deliveryBatch) {
 		if remaining := time.Until(retryDeadline); backoff > remaining {
 			backoff = remaining
 		}
-		c.logf(LogLevelDebug, "retry scheduled: attempts=%d backoff=%s", attempts, backoff)
+		// Retry with a fresh label (<label>_<attempt>), exactly like the Flink
+		// connector: the failed attempt may have registered the old label in
+		// Doris, and re-using it would surface as "Label Already Exists".
+		batch.label = fmt.Sprintf("%s_%d", baseLabel, attempts)
+		c.logf(LogLevelInfo, "stream load failed, to retry: attempt=%d/%d next_label=%s backoff=%s err=%v", attempts, maxRetries+1, batch.label, backoff, err)
 		time.Sleep(backoff)
 	}
 }
diff --git a/stream-load-sdks/doris_go_stream_load/client_test.go b/stream-load-sdks/doris_go_stream_load/client_test.go
index 7f178c4..bdf2c0e 100644
--- a/stream-load-sdks/doris_go_stream_load/client_test.go
+++ b/stream-load-sdks/doris_go_stream_load/client_test.go
@@ -9,6 +9,7 @@ import (
 	"crypto/x509/pkix"
 	"encoding/pem"
 	"errors"
+	"fmt"
 	"io"
 	"log"
 	"math/big"
@@ -1622,10 +1623,9 @@ func TestDialFailureRetriesWithoutPolling(t *testing.T) {
 	if got := sender.PollCalls(); got != 0 {
 		t.Fatalf("poll calls = %d, want 0 (dial failure must not trigger polling)", got)
 	}
-	labels := sender.Labels()
-	if labels[0] != labels[1] {
-		t.Fatalf("labels differ on dial failure retry: %q != %q, want same label (label was never registered)", labels[0], labels[1])
-	}
+	// Every retry uses <label>_<attempt>, like the Flink connector, even when
+	// the previous label was never registered.
+	assertRetryLabels(t, sender.Labels())
 }
 
 func TestLabelAlreadyExistsFinishedTreatedAsSuccess(t *testing.T) {
@@ -1871,22 +1871,29 @@ func TestClassifyResponseError(t *testing.T) {
 		wantRetriable bool
 		wantAmbiguous bool
 	}{
-		// HTTP code classification (no response body)
+		// Like the Flink connector, every non-2xx HTTP status is retried
+		// (bounded by MaxRetries), regardless of the code.
 		{"HTTP 500", 500, nil, true, false},
 		{"HTTP 503", 503, nil, true, false},
 		{"HTTP 429", 429, nil, true, false},
 		{"HTTP 408", 408, nil, true, false},
-		{"HTTP 400", 400, nil, false, false},
-		{"HTTP 401", 401, nil, false, false},
-		{"HTTP 403", 403, nil, false, false},
-		{"HTTP 404", 404, nil, false, false},
-		// Status Fail with various HTTP codes
-		{"Fail 200 not retriable", 200, &StreamLoadResponse{Status: "Fail"}, false, false},
+		{"HTTP 400", 400, nil, true, false},
+		{"HTTP 401", 401, nil, true, false},
+		{"HTTP 403", 403, nil, true, false},
+		{"HTTP 404", 404, nil, true, false},
+		// Doris reports load failures as HTTP 200 + Status Fail. These include
+		// transient errors (too many versions, txn limit, memory limit, ...)
+		// and must be retried with a fresh label.
+		{"Fail 200 retriable", 200, &StreamLoadResponse{Status: "Fail"}, true, false},
+		{"Fail 200 too many versions retriable", 200, &StreamLoadResponse{Status: "Fail", Message: "[INTERNAL_ERROR]tablet error: [E-235]failed to init rowset builder. version count: 2001, exceed limit: 2000"}, true, false},
+		{"Fail 200 txn limit retriable", 200, &StreamLoadResponse{Status: "Fail", Message: "current running txns on db 10003 is 1000, larger than limit 1000"}, true, false},
 		{"Fail 500 retriable", 500, &StreamLoadResponse{Status: "Fail"}, true, false},
-		// Label Already Exists
+		{"unknown status retriable", 200, &StreamLoadResponse{Status: "Whatever"}, true, false},
+		// Label Already Exists: RUNNING is ambiguous (poll the label); anything
+		// else that is not FINISHED is retried with a fresh label.
 		{"Label Already Exists RUNNING ambiguous", 200, &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: "RUNNING"}, false, true},
-		{"Label Already Exists other not ambiguous", 200, &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: "OTHER"}, false, false},
-		{"Label Already Exists empty not ambiguous", 200, &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: ""}, false, false},
+		{"Label Already Exists other retriable", 200, &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: "OTHER"}, true, false},
+		{"Label Already Exists empty retriable", 200, &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: ""}, true, false},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -2027,14 +2034,19 @@ func TestLabelAlreadyExistsRunningTriggersPolling(t *testing.T) {
 	}
 }
 
-func TestLabelAlreadyExistsUnknownExistingStatusFails(t *testing.T) {
-	// An ExistingJobStatus that is neither RUNNING nor FINISHED is a permanent failure.
+func TestLabelAlreadyExistsUnknownExistingStatusRetriesWithFreshLabel(t *testing.T) {
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	// An ExistingJobStatus that is neither RUNNING nor FINISHED is not ambiguous:
+	// like the Flink connector, retry with a fresh label (no polling) until
+	// MaxRetries is exhausted.
+	response := &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: ""}
 	sender := &fakeSender{
 		outcomes: []fakeOutcome{
-			{
-				outcome: sendOutcome{statusCode: 200, response: &StreamLoadResponse{Status: "Label Already Exists", ExistingJobStatus: ""}},
-				err:     &streamLoadError{StatusCode: 200, Message: "label already exists", Retriable: false, Ambiguous: false},
-			},
+			{outcome: sendOutcome{statusCode: 200, response: response}, err: classifyResponseError(200, nil, response)},
+			{outcome: sendOutcome{statusCode: 200, response: response}, err: classifyResponseError(200, nil, response)},
+			{outcome: sendOutcome{statusCode: 200, response: response}, err: classifyResponseError(200, nil, response)},
+			{outcome: sendOutcome{statusCode: 200, response: response}, err: classifyResponseError(200, nil, response)},
+			{outcome: sendOutcome{statusCode: 200, response: response}, err: classifyResponseError(200, nil, response)},
 		},
 	}
 	client := newTestClient(t, sender, Config{
@@ -2054,13 +2066,244 @@ func TestLabelAlreadyExistsUnknownExistingStatusFails(t *testing.T) {
 	}
 	result := handle.Wait()
 	if result.Err == nil {
-		t.Fatal("Wait() error = nil, want failure for Label Already Exists with unknown ExistingJobStatus")
+		t.Fatal("Wait() error = nil, want failure once retries are exhausted")
 	}
 	if got := sender.PollCalls(); got != 0 {
 		t.Fatalf("poll calls = %d, want 0 (non-ambiguous, must not poll)", got)
 	}
+	if got := sender.Attempts(); got != defaultMaxRetries+1 {
+		t.Fatalf("sender attempts = %d, want %d (initial + MaxRetries)", got, defaultMaxRetries+1)
+	}
+	assertRetryLabels(t, sender.Labels())
+}
+
+// assertRetryLabels checks that retries used <base>_<attempt> labels like the
+// Flink connector (label + "_" + retry).
+func assertRetryLabels(t *testing.T, labels []string) {
+	t.Helper()
+	if len(labels) == 0 {
+		t.Fatal("no labels recorded")
+	}
+	for i := 1; i < len(labels); i++ {
+		want := fmt.Sprintf("%s_%d", labels[0], i)
+		if labels[i] != want {
+			t.Fatalf("labels[%d] = %q, want %q", i, labels[i], want)
+		}
+	}
+}
+
+func TestFailStatusRetriedWithFreshLabel(t *testing.T) {
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	// Doris reports "too many versions" (and most other transient failures) as
+	// HTTP 200 + Status Fail. The Flink connector retries these with a new
+	// label; so must we.
+	failResponse := &StreamLoadResponse{
+		Status:  "Fail",
+		Message: "[INTERNAL_ERROR]tablet error: [E-235]failed to init rowset builder. version count: 2001, exceed limit: 2000. tablet: 10042",
+	}
+	sender := &fakeSender{
+		outcomes: []fakeOutcome{
+			{outcome: sendOutcome{statusCode: 200, response: failResponse}, err: classifyResponseError(200, nil, failResponse)},
+			{outcome: sendOutcome{statusCode: 200, response: failResponse}, err: classifyResponseError(200, nil, failResponse)},
+			{outcome: sendOutcome{statusCode: 200, response: &StreamLoadResponse{Status: "Success"}}},
+		},
+	}
+	client := newTestClient(t, sender, Config{
+		StreamLoadURL:      "http://example.invalid/api/db/table/_stream_load",
+		Columns:            []string{"c1"},
+		Mode:               ModeCSV,
+		Linger:             10 * time.Millisecond,
+		DorisUploadWorkers: 1,
+		MaxUploadQueueSize: 8,
+		Logger:             log.New(io.Discard, "", 0),
+	})
+	defer client.Close()
+
+	handle, err := client.SendRecord("a")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	result := handle.Wait()
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after retrying Status Fail", result.Err)
+	}
+	if result.Attempts != 3 {
+		t.Fatalf("attempts = %d, want 3", result.Attempts)
+	}
+	if got := sender.PollCalls(); got != 0 {
+		t.Fatalf("poll calls = %d, want 0", got)
+	}
+	assertRetryLabels(t, sender.Labels())
+}
+
+func TestMaxRetriesBoundsAttempts(t *testing.T) {
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	failResponse := &StreamLoadResponse{Status: "Fail", Message: "current running txns on db 10003 is 1000, larger than limit 1000"}
+	var outcomes []fakeOutcome
+	for i := 0; i < 10; i++ {
+		outcomes = append(outcomes, fakeOutcome{outcome: sendOutcome{statusCode: 200, response: failResponse}, err: classifyResponseError(200, nil, failResponse)})
+	}
+	sender := &fakeSender{outcomes: outcomes}
+	client := newTestClient(t, sender, Config{
+		StreamLoadURL:      "http://example.invalid/api/db/table/_stream_load",
+		Columns:            []string{"c1"},
+		Mode:               ModeCSV,
+		Linger:             10 * time.Millisecond,
+		DorisUploadWorkers: 1,
+		MaxRetries:         2,
+		MaxUploadQueueSize: 8,
+		Logger:             log.New(io.Discard, "", 0),
+	})
+	defer client.Close()
+
+	handle, err := client.SendRecord("a")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	result := handle.Wait()
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure after MaxRetries")
+	}
+	if result.Attempts != 3 {
+		t.Fatalf("attempts = %d, want 3 (initial + 2 retries)", result.Attempts)
+	}
+	if got := sender.Attempts(); got != 3 {
+		t.Fatalf("sender attempts = %d, want 3", got)
+	}
+	if result.Response == nil || result.Response.Status != "Fail" {
+		t.Fatalf("result.Response = %#v, want last Doris Fail response", result.Response)
+	}
+	if !strings.Contains(result.Err.Error(), "larger than limit") {
+		t.Fatalf("result.Err = %v, want last Doris message", result.Err)
+	}
+	assertRetryLabels(t, sender.Labels())
+}
+
+func TestNegativeMaxRetriesDisablesRetry(t *testing.T) {
+	failResponse := &StreamLoadResponse{Status: "Fail", Message: "boom"}
+	sender := &fakeSender{
+		outcomes: []fakeOutcome{
+			{outcome: sendOutcome{statusCode: 200, response: failResponse}, err: classifyResponseError(200, nil, failResponse)},
+		},
+	}
+	client := newTestClient(t, sender, Config{
+		StreamLoadURL:      "http://example.invalid/api/db/table/_stream_load",
+		Columns:            []string{"c1"},
+		Mode:               ModeCSV,
+		Linger:             10 * time.Millisecond,
+		DorisUploadWorkers: 1,
+		MaxRetries:         -1,
+		MaxUploadQueueSize: 8,
+		Logger:             log.New(io.Discard, "", 0),
+	})
+	defer client.Close()
+
+	handle, err := client.SendRecord("a")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	result := handle.Wait()
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure")
+	}
 	if got := sender.Attempts(); got != 1 {
-		t.Fatalf("sender attempts = %d, want 1 (permanent failure, must not retry)", got)
+		t.Fatalf("sender attempts = %d, want 1 (retries disabled)", got)
+	}
+}
+
+func TestMaxRetriesAlsoBoundsAbortedPollRetries(t *testing.T) {
+	withTestPollBackoff(t, time.Millisecond, 2*time.Millisecond)
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	// Ambiguous send -> ABORTED poll is a retry like any other and counts
+	// against MaxRetries.
+	var outcomes []fakeOutcome
+	var polls []fakePollResponse
+	for i := 0; i < 10; i++ {
+		outcomes = append(outcomes, fakeOutcome{err: &streamLoadError{Message: "timeout", Ambiguous: true}})
+		polls = append(polls, fakePollResponse{state: loadStateResponse{StatusCode: 200, State: "ABORTED"}})
+	}
+	sender := &fakeSender{outcomes: outcomes, pollResponses: polls}
+	client := newTestClient(t, sender, Config{
+		StreamLoadURL:      "http://example.invalid/api/db/table/_stream_load",
+		Columns:            []string{"c1"},
+		Mode:               ModeCSV,
+		Linger:             10 * time.Millisecond,
+		DorisUploadWorkers: 1,
+		MaxRetries:         1,
+		MaxUploadQueueSize: 8,
+		Logger:             log.New(io.Discard, "", 0),
+	})
+	defer client.Close()
+
+	handle, err := client.SendRecord("a")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	result := handle.Wait()
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure after MaxRetries")
+	}
+	if got := sender.Attempts(); got != 2 {
+		t.Fatalf("sender attempts = %d, want 2 (initial + 1 retry)", got)
+	}
+	if got := sender.PollCalls(); got != 2 {
+		t.Fatalf("poll calls = %d, want 2", got)
+	}
+	assertRetryLabels(t, sender.Labels())
+}
+
+func TestHTTPSenderRetriesStatusFailWithFreshLabel(t *testing.T) {
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	// End-to-end through the real httpSender: Doris answers HTTP 200 with
+	// Status Fail (too many versions), then succeeds. The retry must carry a
+	// new label header.
+	transport := &sequentialTransport{
+		responses: []*http.Response{
+			{
+				StatusCode: http.StatusOK,
+				Header:     make(http.Header),
+				Body:       io.NopCloser(strings.NewReader(`{"TxnId":1,"Label":"x","Status":"Fail","Message":"[INTERNAL_ERROR]tablet error: [E-235]failed to init rowset builder. version count: 2001, exceed limit: 2000"}`)),
+			},
+			{
+				StatusCode: http.StatusOK,
+				Header:     make(http.Header),
+				Body:       io.NopCloser(strings.NewReader(`{"TxnId":2,"Label":"x_1","Status":"Success","NumberLoadedRows":1}`)),
+			},
+		},
+	}
+	client, err := NewClient(Config{
+		StreamLoadURL:      "http://example.invalid/api/db/table/_stream_load",
+		Columns:            []string{"c1"},
+		Mode:               ModeCSV,
+		Linger:             10 * time.Millisecond,
+		DorisUploadWorkers: 1,
+		MaxUploadQueueSize: 8,
+		Logger:             log.New(io.Discard, "", 0),
+		HTTPClient:         &http.Client{Transport: transport, Timeout: time.Second},
+	})
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	defer client.Close()
+
+	handle, err := client.SendRecord("a")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	result := handle.Wait()
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after retry", result.Err)
+	}
+	if result.Attempts != 2 {
+		t.Fatalf("attempts = %d, want 2", result.Attempts)
+	}
+	requests := transport.AllRequests()
+	if len(requests) != 2 {
+		t.Fatalf("requests = %d, want 2", len(requests))
+	}
+	first, second := requests[0].Header.Get("label"), requests[1].Header.Get("label")
+	if first == "" || second != first+"_1" {
+		t.Fatalf("labels = %q, %q; want second label = first + \"_1\"", first, second)
 	}
 }
 
diff --git a/stream-load-sdks/doris_go_stream_load/config.go b/stream-load-sdks/doris_go_stream_load/config.go
index 03b51ed..6920067 100644
--- a/stream-load-sdks/doris_go_stream_load/config.go
+++ b/stream-load-sdks/doris_go_stream_load/config.go
@@ -21,6 +21,7 @@ const (
 	defaultMaxUploadQueueSize        = 1
 	defaultBatchBytes                = maxBatchBytes
 	defaultDorisUploadWorkers        = 1
+	defaultMaxRetries                = 3
 	defaultLinger                    = 5 * time.Millisecond
 	defaultDorisUploadTimeout        = 300 * time.Second
 	defaultDorisUploadRequestTimeout = 300 * time.Second
@@ -93,13 +94,20 @@ type Config struct {
 
 	DorisUploadTimeout        time.Duration
 	DorisUploadRequestTimeout time.Duration
-	CallbackTimeout           time.Duration
-	SlowCallbackWarn          time.Duration
-	LabelPrefix               string
-	StatusPollTimeout         time.Duration
-	FakeSend                  bool
-	FakeSendDelay             time.Duration
-	FakeSendDelaySet          bool
+	// MaxRetries is the number of additional upload attempts made after a
+	// failed stream load before the batch is reported as failed, so a batch is
+	// sent at most MaxRetries+1 times. Every retry uses a fresh label
+	// (<label>_<attempt>). This mirrors the Flink connector's sink.max-retries
+	// (default 3). Zero selects the default; a negative value disables retries.
+	// DorisUploadTimeout still bounds the total time spent retrying.
+	MaxRetries        int
+	CallbackTimeout   time.Duration
+	SlowCallbackWarn  time.Duration
+	LabelPrefix       string
+	StatusPollTimeout time.Duration
+	FakeSend          bool
+	FakeSendDelay     time.Duration
+	FakeSendDelaySet  bool
 
 	CSVSeparator string
 	CSVQuote     string
@@ -138,6 +146,9 @@ func (c Config) withDefaults() Config {
 	if c.DorisUploadRequestTimeout == 0 {
 		c.DorisUploadRequestTimeout = defaultDorisUploadRequestTimeout
 	}
+	if c.MaxRetries == 0 {
+		c.MaxRetries = defaultMaxRetries
+	}
 	if c.CallbackTimeout <= 0 {
 		c.CallbackTimeout = defaultCallbackTimeout
 	}
@@ -177,6 +188,14 @@ func (c Config) withDefaults() Config {
 	return c
 }
 
+// maxRetries returns the effective retry count: negative values disable retries.
+func (c Config) maxRetries() int {
+	if c.MaxRetries < 0 {
+		return 0
+	}
+	return c.MaxRetries
+}
+
 func (c Config) validate() error {
 	if len(c.Columns) == 0 {
 		return errors.New("columns must be configured")
diff --git a/stream-load-sdks/doris_go_stream_load/loader_config.go b/stream-load-sdks/doris_go_stream_load/loader_config.go
index 08e9d22..7968165 100644
--- a/stream-load-sdks/doris_go_stream_load/loader_config.go
+++ b/stream-load-sdks/doris_go_stream_load/loader_config.go
@@ -30,6 +30,7 @@ type LoaderConfig struct {
 
 	DorisUploadTimeout        string `json:"doris_upload_timeout,omitempty"`
 	DorisUploadRequestTimeout string `json:"doris_upload_request_timeout,omitempty"`
+	MaxRetries                int    `json:"max_retries,omitempty"`
 	CallbackTimeout           string `json:"callback_timeout,omitempty"`
 	SlowCallbackWarn          string `json:"slow_callback_warn,omitempty"`
 	LabelPrefix               string `json:"label_prefix,omitempty"`
@@ -89,6 +90,7 @@ func LoaderConfigFromConfig(cfg Config) LoaderConfig {
 		Validation:                cfg.Validation,
 		DorisUploadTimeout:        formatDuration(cfg.DorisUploadTimeout),
 		DorisUploadRequestTimeout: formatDuration(cfg.DorisUploadRequestTimeout),
+		MaxRetries:                cfg.MaxRetries,
 		CallbackTimeout:           formatDuration(cfg.CallbackTimeout),
 		SlowCallbackWarn:          formatDuration(cfg.SlowCallbackWarn),
 		LabelPrefix:               cfg.LabelPrefix,
@@ -120,6 +122,7 @@ func (lc LoaderConfig) Config() (Config, error) {
 		MaxUploadQueueSize:  lc.MaxUploadQueueSize,
 		BatchBytes:          lc.BatchBytes,
 		DorisUploadWorkers:  lc.DorisUploadWorkers,
+		MaxRetries:          lc.MaxRetries,
 		Validation:          lc.Validation,
 		LabelPrefix:         lc.LabelPrefix,
 		CSVSeparator:        lc.CSVSeparator,
diff --git a/stream-load-sdks/doris_go_stream_load/loader_config_test.go b/stream-load-sdks/doris_go_stream_load/loader_config_test.go
index 2a6aefc..939f5ab 100644
--- a/stream-load-sdks/doris_go_stream_load/loader_config_test.go
+++ b/stream-load-sdks/doris_go_stream_load/loader_config_test.go
@@ -18,6 +18,7 @@ func TestLoaderConfigConvertsToRuntimeConfig(t *testing.T) {
 		MaxQueueWaitTime:          "2s",
 		DorisUploadTimeout:        "100s",
 		DorisUploadRequestTimeout: "30s",
+		MaxRetries:                5,
 		FakeSend:                  true,
 		FakeSendDelay:             "500ms",
 		Validation:                ValidateSyntax,
@@ -43,6 +44,9 @@ func TestLoaderConfigConvertsToRuntimeConfig(t *testing.T) {
 	if cfg.DorisUploadRequestTimeout != 30*time.Second {
 		t.Fatalf("DorisUploadRequestTimeout = %s, want 30s", cfg.DorisUploadRequestTimeout)
 	}
+	if cfg.MaxRetries != 5 {
+		t.Fatalf("MaxRetries = %d, want 5", cfg.MaxRetries)
+	}
 	if !cfg.FakeSend {
 		t.Fatal("FakeSend = false, want true")
 	}
diff --git a/stream-load-sdks/doris_go_stream_load/mock_backend_test.go b/stream-load-sdks/doris_go_stream_load/mock_backend_test.go
new file mode 100644
index 0000000..63b3543
--- /dev/null
+++ b/stream-load-sdks/doris_go_stream_load/mock_backend_test.go
@@ -0,0 +1,685 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dorisstreamload
+
+// End-to-end tests of the retry / ambiguity-resolution policy against a mock
+// Doris backend. They drive the real httpSender (HTTP, headers, redirects,
+// label polling) rather than the in-process fakeSender, and each test states
+// how the Flink connector (DorisBatchStreamLoad.load) behaves in the same
+// situation so the parity — and the deliberate improvements — are explicit.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// mockLoad is one _stream_load request as seen by the mock backend.
+type mockLoad struct {
+	label string
+	body  string
+	auth  string
+}
+
+// mockLoadHandler scripts the backend's reaction to one _stream_load attempt.
+type mockLoadHandler func(w http.ResponseWriter, load mockLoad)
+
+// mockDoris is a scripted stand-in for a Doris FE/BE. It serves
+// PUT /api/{db}/{table}/_stream_load according to `script` (one handler per
+// attempt, the last one repeating) and GET /api/{db}/get_load_state?label=...
+// according to `states` (one state per poll, the last one repeating).
+type mockDoris struct {
+	t      *testing.T
+	server *httptest.Server
+
+	mu     sync.Mutex
+	script []mockLoadHandler
+	states map[string][]string
+	loads  []mockLoad
+	polls  []string
+}
+
+func newMockDoris(t *testing.T, script ...mockLoadHandler) *mockDoris {
+	t.Helper()
+	m := &mockDoris{t: t, script: script, states: map[string][]string{}}
+	m.server = httptest.NewServer(http.HandlerFunc(m.serve))
+	t.Cleanup(m.server.Close)
+	return m
+}
+
+func (m *mockDoris) streamLoadURL() string {
+	return m.server.URL + "/api/db/table/_stream_load"
+}
+
+// setStates scripts the get_load_state answers for a label ("*" matches any
+// label); the last state repeats for every further poll.
+func (m *mockDoris) setStates(label string, states ...string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.states[label] = states
+}
+
+func (m *mockDoris) serve(w http.ResponseWriter, r *http.Request) {
+	switch {
+	case strings.HasSuffix(r.URL.Path, "/_stream_load"):
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			m.t.Errorf("mock doris: read body: %v", err)
+			return
+		}
+		load := mockLoad{label: r.Header.Get("label"), body: string(body), auth: r.Header.Get("Authorization")}
+		m.mu.Lock()
+		m.loads = append(m.loads, load)
+		idx := len(m.loads) - 1
+		var handler mockLoadHandler
+		if len(m.script) > 0 {
+			if idx >= len(m.script) {
+				idx = len(m.script) - 1
+			}
+			handler = m.script[idx]
+		}
+		m.mu.Unlock()
+		if handler == nil {
+			handler = loadSuccess
+		}
+		handler(w, load)
+	case strings.HasSuffix(r.URL.Path, "/get_load_state"):
+		label := r.URL.Query().Get("label")
+		m.mu.Lock()
+		m.polls = append(m.polls, label)
+		state := "UNKNOWN"
+		key := label
+		if _, ok := m.states[key]; !ok {
+			key = "*"
+		}
+		if states := m.states[key]; len(states) > 0 {
+			state = states[0]
+			if len(states) > 1 {
+				m.states[key] = states[1:]
+			}
+		}
+		m.mu.Unlock()
+		writeJSON(w, http.StatusOK, map[string]any{"msg": "success", "code": 0, "data": state, "count": 0})
+	default:
+		http.NotFound(w, r)
+	}
+}
+
+func (m *mockDoris) Loads() []mockLoad {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return append([]mockLoad(nil), m.loads...)
+}
+
+func (m *mockDoris) Polls() []string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return append([]string(nil), m.polls...)
+}
+
+func (m *mockDoris) labels() []string {
+	loads := m.Loads()
+	labels := make([]string, 0, len(loads))
+	for _, l := range loads {
+		labels = append(labels, l.label)
+	}
+	return labels
+}
+
+func writeJSON(w http.ResponseWriter, status int, payload any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(payload)
+}
+
+// --- scripted backend reactions -------------------------------------------
+
+func loadSuccess(w http.ResponseWriter, load mockLoad) {
+	writeJSON(w, http.StatusOK, map[string]any{
+		"TxnId": 1, "Label": load.label, "Status": "Success", "Message": "OK",
+		"NumberTotalRows": 1, "NumberLoadedRows": 1,
+	})
+}
+
+// loadFail mimics how Doris reports load failures: HTTP 200 + Status Fail.
+func loadFail(message string) mockLoadHandler {
+	return func(w http.ResponseWriter, load mockLoad) {
+		writeJSON(w, http.StatusOK, map[string]any{
+			"TxnId": 1, "Label": load.label, "Status": "Fail", "Message": message,
+			"ErrorURL": "http://be:8040/api/_load_error_log?file=__shard_0/error_log_x",
+		})
+	}
+}
+
+func loadHTTPStatus(status int, body string) mockLoadHandler {
+	return func(w http.ResponseWriter, _ mockLoad) {
+		w.Header().Set("Content-Type", "text/html")
+		w.WriteHeader(status)
+		_, _ = io.WriteString(w, body)
+	}
+}
+
+func loadLabelAlreadyExists(existingJobStatus string) mockLoadHandler {
+	return func(w http.ResponseWriter, load mockLoad) {
+		writeJSON(w, http.StatusOK, map[string]any{
+			"TxnId": -1, "Label": load.label, "Status": "Label Already Exists",
+			"ExistingJobStatus": existingJobStatus,
+			"Message":           fmt.Sprintf("errCode = 2, detailMessage = Label [%s] has already been used, relate to txn [42]", load.label),
+		})
+	}
+}
+
+// loadDropConnection reads the whole request, then closes the TCP connection
+// without answering: the client cannot know whether Doris committed the data.
+func loadDropConnection(w http.ResponseWriter, _ mockLoad) {
+	hj, ok := w.(http.Hijacker)
+	if !ok {
+		panic("mock doris: response writer is not a Hijacker")
+	}
+	conn, _, err := hj.Hijack()
+	if err != nil {
+		panic(err)
+	}
+	_ = conn.Close()
+}
+
+// --- helpers --------------------------------------------------------------
+
+const tooManyVersionsMsg = "[INTERNAL_ERROR]tablet error: [E-235]failed to init rowset builder. version count: 2001, exceed limit: 2000. tablet: 10042"
+const txnLimitMsg = "current running txns on db 10003 is 1000, larger than limit 1000"
+
+func newMockClient(t *testing.T, m *mockDoris, mutate func(*Config)) *Client {
+	t.Helper()
+	withTestUploadRetryBackoff(t, time.Millisecond, 2*time.Millisecond)
+	withTestPollBackoff(t, time.Millisecond, 2*time.Millisecond)
+	cfg := Config{
+		StreamLoadURL:       m.streamLoadURL(),
+		Columns:             []string{"c1"},
+		Mode:                ModeCSV,
+		Linger:              5 * time.Millisecond,
+		DorisUploadWorkers:  1,
+		MaxUploadQueueSize:  8,
+		AuthenticationType:  AuthenticationBasic,
+		AuthenticationToken: "root:secret",
+		StatusPollTimeout:   2 * time.Second,
+		Logger:              log.New(io.Discard, "", 0),
+	}
+	if mutate != nil {
+		mutate(&cfg)
+	}
+	client, err := NewClient(cfg)
+	if err != nil {
+		t.Fatalf("NewClient() error = %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+	return client
+}
+
+func sendOne(t *testing.T, client *Client) DeliveryResult {
+	t.Helper()
+	handle, err := client.SendRecord("row-1")
+	if err != nil {
+		t.Fatalf("SendRecord() error = %v", err)
+	}
+	return handle.Wait()
+}
+
+func assertSameBodies(t *testing.T, loads []mockLoad) {
+	t.Helper()
+	for i := 1; i < len(loads); i++ {
+		if loads[i].body != loads[0].body {
+			t.Fatalf("retry %d re-sent a different body: %q vs %q", i, loads[i].body, loads[0].body)
+		}
+	}
+}
+
+// --- Doris-level failures: retried with a fresh label (same as Flink) ------
+
+func TestMockDorisTooManyVersionsIsRetriedWithFreshLabel(t *testing.T) {
+	// Flink: HTTP 200 + Status Fail -> DorisBatchLoadException -> retried with
+	// label_N. Before this change the Go SDK gave up immediately.
+	m := newMockDoris(t, loadFail(tooManyVersionsMsg), loadFail(tooManyVersionsMsg), loadSuccess)
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after retries", result.Err)
+	}
+	if result.Attempts != 3 {
+		t.Fatalf("attempts = %d, want 3", result.Attempts)
+	}
+	loads := m.Loads()
+	if len(loads) != 3 {
+		t.Fatalf("backend saw %d loads, want 3", len(loads))
+	}
+	assertRetryLabels(t, m.labels())
+	assertSameBodies(t, loads)
+	if got := m.Polls(); len(got) != 0 {
+		t.Fatalf("polls = %v, want none (definitive Fail is not ambiguous)", got)
+	}
+	for i, l := range loads {
+		if l.auth == "" {
+			t.Fatalf("load %d lost basic auth on retry", i)
+		}
+	}
+}
+
+func TestMockDorisTxnLimitExhaustsMaxRetries(t *testing.T) {
+	// Flink: retry until sink.max-retries then throw with the last message.
+	m := newMockDoris(t, loadFail(txnLimitMsg))
+	client := newMockClient(t, m, func(c *Config) { c.MaxRetries = 2 })
+
+	result := sendOne(t, client)
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure after MaxRetries")
+	}
+	if result.Attempts != 3 {
+		t.Fatalf("attempts = %d, want 3 (1 + MaxRetries)", result.Attempts)
+	}
+	if got := len(m.Loads()); got != 3 {
+		t.Fatalf("backend saw %d loads, want 3", got)
+	}
+	assertRetryLabels(t, m.labels())
+	if result.Response == nil || result.Response.Status != "Fail" {
+		t.Fatalf("result.Response = %+v, want last Doris Fail response", result.Response)
+	}
+	msg := result.Err.Error()
+	if !strings.Contains(msg, txnLimitMsg) || !strings.Contains(msg, "see more in http://be:8040") {
+		t.Fatalf("error = %q, want Doris message and ErrorURL", msg)
+	}
+}
+
+func TestMockDorisNon2xxIsRetried(t *testing.T) {
+	// Flink: "stream load failed with {}, reason {}, to retry".
+	m := newMockDoris(t, loadHTTPStatus(http.StatusServiceUnavailable, "<html>be busy</html>"), loadHTTPStatus(http.StatusBadGateway, ""), loadSuccess)
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success", result.Err)
+	}
+	if got := len(m.Loads()); got != 3 {
+		t.Fatalf("backend saw %d loads, want 3", got)
+	}
+	assertRetryLabels(t, m.labels())
+}
+
+func TestMockDorisNegativeMaxRetriesFailsFast(t *testing.T) {
+	m := newMockDoris(t, loadFail(tooManyVersionsMsg), loadSuccess)
+	client := newMockClient(t, m, func(c *Config) { c.MaxRetries = -1 })
+
+	result := sendOne(t, client)
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure with retries disabled")
+	}
+	if got := len(m.Loads()); got != 1 {
+		t.Fatalf("backend saw %d loads, want 1", got)
+	}
+}
+
+func TestMockDorisUploadTimeoutBoundsRetries(t *testing.T) {
+	// DorisUploadTimeout caps the retry window even when MaxRetries is large.
+	m := newMockDoris(t, loadFail(txnLimitMsg))
+	client := newMockClient(t, m, func(c *Config) {
+		c.MaxRetries = 1000
+		c.DorisUploadTimeout = 30 * time.Millisecond
+	})
+
+	result := sendOne(t, client)
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want failure")
+	}
+	if !strings.Contains(result.Err.Error(), "doris upload timeout") {
+		t.Fatalf("error = %v, want upload timeout", result.Err)
+	}
+	if got := len(m.Loads()); got < 2 || got > 200 {
+		t.Fatalf("backend saw %d loads, want a handful bounded by the 30ms budget", got)
+	}
+}
+
+// --- connection refused: retried, no polling (same as Flink) ---------------
+
+func TestMockDorisConnectionRefusedIsRetriedWithoutPolling(t *testing.T) {
+	m := newMockDoris(t, loadSuccess)
+
+	// The first dial goes to a closed port; later dials reach the mock.
+	closed, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	closedAddr := closed.Addr().String()
+	_ = closed.Close()
+
+	var dials int32
+	var dialMu sync.Mutex
+	transport := &http.Transport{
+		DisableKeepAlives: true,
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			dialMu.Lock()
+			dials++
+			first := dials == 1
+			dialMu.Unlock()
+			if first {
+				addr = closedAddr
+			}
+			return (&net.Dialer{}).DialContext(ctx, network, addr)
+		},
+	}
+	client := newMockClient(t, m, func(c *Config) { c.HTTPClient = &http.Client{Transport: transport} })
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after dial retry", result.Err)
+	}
+	if result.Attempts != 2 {
+		t.Fatalf("attempts = %d, want 2", result.Attempts)
+	}
+	loads := m.Loads()
+	if len(loads) != 1 {
+		t.Fatalf("backend saw %d loads, want 1 (first attempt never connected)", len(loads))
+	}
+	if !strings.HasSuffix(loads[0].label, "_1") {
+		t.Fatalf("label = %q, want retry suffix _1", loads[0].label)
+	}
+	if got := m.Polls(); len(got) != 0 {
+		t.Fatalf("polls = %v, want none (request never reached Doris)", got)
+	}
+}
+
+// --- ambiguous outcomes: resolved by polling (better than Flink batch) -----
+
+func TestMockDorisDroppedResponseAfterCommitIsNotResent(t *testing.T) {
+	// Doris committed the txn but the response was lost. Flink batch mode
+	// re-sends under label_1 and writes the data twice; the Go SDK polls
+	// get_load_state, sees VISIBLE and reports success without re-sending.
+	m := newMockDoris(t, loadDropConnection, loadSuccess)
+	m.setStates("*", "VISIBLE")
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success resolved via poll", result.Err)
+	}
+	if result.Attempts != 1 {
+		t.Fatalf("attempts = %d, want 1 (no re-send)", result.Attempts)
+	}
+	loads := m.Loads()
+	if len(loads) != 1 {
+		t.Fatalf("backend saw %d loads, want 1 — data must not be written twice", len(loads))
+	}
+	polls := m.Polls()
+	if len(polls) == 0 || polls[0] != loads[0].label {
+		t.Fatalf("polls = %v, want poll of label %q", polls, loads[0].label)
+	}
+	if result.Response == nil || result.Response.Status != "Success" || result.Response.Message != "VISIBLE" {
+		t.Fatalf("result.Response = %+v, want synthesized Success/VISIBLE", result.Response)
+	}
+}
+
+func TestMockDorisDroppedResponseAbortedIsRetriedWithFreshLabel(t *testing.T) {
+	// Doris aborted the txn (nothing loaded) -> safe to retry with a new label.
+	m := newMockDoris(t, loadDropConnection, loadSuccess)
+	m.setStates("*", "ABORTED")
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after ABORTED retry", result.Err)
+	}
+	if result.Attempts != 2 {
+		t.Fatalf("attempts = %d, want 2", result.Attempts)
+	}
+	loads := m.Loads()
+	if len(loads) != 2 {
+		t.Fatalf("backend saw %d loads, want 2", len(loads))
+	}
+	assertRetryLabels(t, m.labels())
+	assertSameBodies(t, loads)
+	if polls := m.Polls(); len(polls) != 1 || polls[0] != loads[0].label {
+		t.Fatalf("polls = %v, want exactly one poll of %q", polls, loads[0].label)
+	}
+}
+
+func TestMockDorisDroppedResponseUnknownLabelIsRetried(t *testing.T) {
+	// Label never registered (request died before FE begin_txn) -> retry.
+	m := newMockDoris(t, loadDropConnection, loadSuccess)
+	// default poll state is UNKNOWN
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success", result.Err)
+	}
+	if got := len(m.Loads()); got != 2 {
+		t.Fatalf("backend saw %d loads, want 2", got)
+	}
+	assertRetryLabels(t, m.labels())
+}
+
+func TestMockDorisDroppedResponsePendingStatesArePolledUntilVisible(t *testing.T) {
+	// Txn still in PREPARE/PRECOMMITTED: keep polling, never re-send.
+	m := newMockDoris(t, loadDropConnection, loadSuccess)
+	m.setStates("*", "PREPARE", "PREPARE", "PRECOMMITTED", "COMMITTED")
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success once COMMITTED", result.Err)
+	}
+	if got := len(m.Loads()); got != 1 {
+		t.Fatalf("backend saw %d loads, want 1", got)
+	}
+	if got := len(m.Polls()); got < 4 {
+		t.Fatalf("polls = %d, want at least 4 (PREPARE, PREPARE, PRECOMMITTED, COMMITTED)", got)
+	}
+}
+
+func TestMockDorisPollTimeoutFailsWithoutResend(t *testing.T) {
+	// Label never reaches a terminal state: report failure, never re-send —
+	// the data may be in, and retrying could write it twice.
+	m := newMockDoris(t, loadDropConnection, loadSuccess)
+	m.setStates("*", "PREPARE")
+	client := newMockClient(t, m, func(c *Config) { c.StatusPollTimeout = 50 * time.Millisecond })
+
+	result := sendOne(t, client)
+	if result.Err == nil {
+		t.Fatal("Wait() error = nil, want poll timeout failure")
+	}
+	if !strings.Contains(result.Err.Error(), "poll timeout") {
+		t.Fatalf("error = %v, want poll timeout", result.Err)
+	}
+	if got := len(m.Loads()); got != 1 {
+		t.Fatalf("backend saw %d loads, want 1 (never re-send an unresolved label)", got)
+	}
+	if got := len(m.Polls()); got == 0 {
+		t.Fatal("expected label polls before giving up")
+	}
+}
+
+// --- Label Already Exists --------------------------------------------------
+
+func TestMockDorisLabelAlreadyExistsFinishedIsSuccess(t *testing.T) {
+	// Flink batch mode retries this with label_1 and duplicates the data.
+	m := newMockDoris(t, loadLabelAlreadyExists("FINISHED"), loadSuccess)
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success (data already loaded)", result.Err)
+	}
+	if got := len(m.Loads()); got != 1 {
+		t.Fatalf("backend saw %d loads, want 1", got)
+	}
+	if got := m.Polls(); len(got) != 0 {
+		t.Fatalf("polls = %v, want none", got)
+	}
+}
+
+func TestMockDorisLabelAlreadyExistsRunningIsPolledThenSuccess(t *testing.T) {
+	// The earlier request with this label is still in flight: wait for it.
+	m := newMockDoris(t, loadLabelAlreadyExists("RUNNING"), loadSuccess)
+	m.setStates("*", "PREPARE", "VISIBLE")
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success via poll", result.Err)
+	}
+	if got := len(m.Loads()); got != 1 {
+		t.Fatalf("backend saw %d loads, want 1 (must not re-send while RUNNING)", got)
+	}
+	if got := len(m.Polls()); got < 2 {
+		t.Fatalf("polls = %d, want at least 2", got)
+	}
+}
+
+func TestMockDorisLabelAlreadyExistsRunningThenAbortedIsRetried(t *testing.T) {
+	m := newMockDoris(t, loadLabelAlreadyExists("RUNNING"), loadSuccess)
+	m.setStates("*", "ABORTED")
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after retry", result.Err)
+	}
+	if got := len(m.Loads()); got != 2 {
+		t.Fatalf("backend saw %d loads, want 2", got)
+	}
+	assertRetryLabels(t, m.labels())
+}
+
+func TestMockDorisLabelAlreadyExistsUnknownStatusIsRetriedWithFreshLabel(t *testing.T) {
+	// Neither FINISHED nor RUNNING: not ambiguous, retry like Flink.
+	m := newMockDoris(t, loadLabelAlreadyExists(""), loadSuccess)
+	client := newMockClient(t, m, nil)
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success after retry", result.Err)
+	}
+	if got := len(m.Loads()); got != 2 {
+		t.Fatalf("backend saw %d loads, want 2", got)
+	}
+	if got := m.Polls(); len(got) != 0 {
+		t.Fatalf("polls = %v, want none", got)
+	}
+	assertRetryLabels(t, m.labels())
+}
+
+// --- FE 307 redirect to BE --------------------------------------------------
+
+func TestMockDorisRetryAfterFERedirectUsesFreshLabelAndKeepsAuth(t *testing.T) {
+	// Real deployments hit the FE, which 307-redirects each attempt to a BE.
+	// Every retry must go back through the FE with a new label and its auth.
+	be := newMockDoris(t, loadFail(tooManyVersionsMsg), loadSuccess)
+	var feMu sync.Mutex
+	var feHits int
+	fe := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/_stream_load") {
+			feMu.Lock()
+			feHits++
+			feMu.Unlock()
+			http.Redirect(w, r, be.server.URL+r.URL.Path, http.StatusTemporaryRedirect)
+			return
+		}
+		// get_load_state is served by the FE itself.
+		be.serve(w, r)
+	}))
+	t.Cleanup(fe.Close)
+
+	client := newMockClient(t, be, func(c *Config) { c.StreamLoadURL = fe.URL + "/api/db/table/_stream_load" })
+
+	result := sendOne(t, client)
+	if result.Err != nil {
+		t.Fatalf("Wait() error = %v, want success", result.Err)
+	}
+	feMu.Lock()
+	hits := feHits
+	feMu.Unlock()
+	if hits != 2 {
+		t.Fatalf("FE saw %d stream load requests, want 2", hits)
+	}
+	loads := be.Loads()
+	if len(loads) != 2 {
+		t.Fatalf("BE saw %d loads, want 2", len(loads))
+	}
+	assertRetryLabels(t, be.labels())
+	assertSameBodies(t, loads)
+	for i, l := range loads {
+		if !strings.HasPrefix(l.auth, "Basic ") {
+			t.Fatalf("BE load %d lost basic auth across redirect: %q", i, l.auth)
+		}
+	}
+}
+
+// --- several batches in flight keep independent retry state ----------------
+
+func TestMockDorisConcurrentBatchesRetryIndependently(t *testing.T) {
+	// Two workers, two batches: one fails once then succeeds, the other
+	// succeeds outright. Each handle must see its own attempt count.
+	var mu sync.Mutex
+	failed := map[string]bool{}
+	m := newMockDoris(t, func(w http.ResponseWriter, load mockLoad) {
+		mu.Lock()
+		shouldFail := strings.Contains(load.body, "flaky") && !failed[strings.TrimSuffix(load.label, "_1")]
+		if shouldFail {
+			failed[load.label] = true
+		}
+		mu.Unlock()
+		if shouldFail {
+			loadFail(tooManyVersionsMsg)(w, load)
+			return
+		}
+		loadSuccess(w, load)
+	})
+	client := newMockClient(t, m, func(c *Config) {
+		c.DorisUploadWorkers = 2
+		c.BatchBytes = 8 // force one record per batch
+	})
+
+	hFlaky, err := client.SendRecord("flaky")
+	if err != nil {
+		t.Fatal(err)
+	}
+	hOK, err := client.SendRecord("fine")
+	if err != nil {
+		t.Fatal(err)
+	}
+	rFlaky, rOK := hFlaky.Wait(), hOK.Wait()
+	if rFlaky.Err != nil || rOK.Err != nil {
+		t.Fatalf("errors: flaky=%v ok=%v", rFlaky.Err, rOK.Err)
+	}
+	if rFlaky.Attempts != 2 {
+		t.Fatalf("flaky attempts = %d, want 2", rFlaky.Attempts)
+	}
+	if rOK.Attempts != 1 {
+		t.Fatalf("ok attempts = %d, want 1", rOK.Attempts)
+	}
+	if got := len(m.Loads()); got != 3 {
+		t.Fatalf("backend saw %d loads, want 3", got)
+	}
+}
diff --git a/stream-load-sdks/doris_go_stream_load/sender.go b/stream-load-sdks/doris_go_stream_load/sender.go
index 42fa0b7..ea24243 100644
--- a/stream-load-sdks/doris_go_stream_load/sender.go
+++ b/stream-load-sdks/doris_go_stream_load/sender.go
@@ -105,14 +105,28 @@ type streamLoadError struct {
 }
 
 func (e *streamLoadError) Error() string {
+	var b strings.Builder
+	fmt.Fprintf(&b, "stream load failed: http=%d", e.StatusCode)
+	if e.Response != nil && e.Response.Status != "" {
+		fmt.Fprintf(&b, " status=%s", e.Response.Status)
+	}
+	respMsg := ""
+	if e.Response != nil {
+		respMsg = e.Response.Message
+	}
 	switch {
-	case e.Response != nil && e.Response.Status != "":
-		return fmt.Sprintf("stream load failed: http=%d status=%s message=%s", e.StatusCode, e.Response.Status, e.Response.Message)
+	case e.Message != "" && respMsg != "" && e.Message != respMsg:
+		// e.g. an upload-timeout error wrapping the last Doris response.
+		fmt.Fprintf(&b, " message=%s (last response: %s)", e.Message, respMsg)
 	case e.Message != "":
-		return fmt.Sprintf("stream load failed: http=%d message=%s", e.StatusCode, e.Message)
-	default:
-		return fmt.Sprintf("stream load failed: http=%d", e.StatusCode)
+		fmt.Fprintf(&b, " message=%s", e.Message)
+	case respMsg != "":
+		fmt.Fprintf(&b, " message=%s", respMsg)
 	}
+	if e.Response != nil && e.Response.ErrorURL != "" {
+		b.WriteString(", see more in " + e.Response.ErrorURL)
+	}
+	return b.String()
 }
 
 func (e *streamLoadError) Is(target error) bool {
@@ -338,46 +352,43 @@ func classifyTransportError(err error) error {
 	return &streamLoadError{StatusCode: 0, Message: err.Error(), Retriable: false, Ambiguous: true}
 }
 
+// classifyResponseError decides what to do with a stream load response that is
+// not a success. It follows the Flink connector (DorisBatchStreamLoad.load):
+// every failed load — a non-2xx HTTP status or a 2xx body whose Status is not
+// Success/Publish Timeout — is retried with a fresh label, bounded by
+// Config.MaxRetries and Config.DorisUploadTimeout.
+//
+// Doris reports almost all load failures as HTTP 200 with Status "Fail", and
+// many of them are transient: "[E-235] too many versions", "[E-238] too many
+// segments", "current running txns on db ... is larger than limit", memory
+// limit exceeded, txn/publish timeouts, no available backend, and so on. The
+// HTTP status code therefore carries no information about retriability and
+// is deliberately not consulted.
+//
+// The only outcome that is not retried immediately is "Label Already Exists"
+// with ExistingJobStatus RUNNING: the earlier request with this label is still
+// in flight, so the result is ambiguous and the label is polled instead of
+// re-sending the data. (FINISHED is treated as success by isHTTPSuccess.)
 func classifyResponseError(statusCode int, body []byte, response *StreamLoadResponse) error {
 	message := strings.TrimSpace(string(body))
 	if response != nil && response.Message != "" {
 		message = response.Message
 	}
 
-	retriable := false
-	switch {
-	case statusCode == http.StatusTooManyRequests:
-		retriable = true
-	case statusCode == http.StatusRequestTimeout:
-		retriable = true
-	case statusCode >= 500:
-		retriable = true
-	}
-
-	if response != nil {
-		switch response.Status {
-		case "Success", "Publish Timeout":
-			retriable = false
-		case "Label Already Exists":
-			retriable = false
-			return &streamLoadError{
-				StatusCode: statusCode,
-				Message:    message,
-				Retriable:  false,
-				Ambiguous:  strings.EqualFold(response.ExistingJobStatus, "RUNNING"),
-				Response:   response,
-			}
-		case "Fail":
-			if statusCode >= 500 {
-				retriable = true
-			}
+	if response != nil && response.Status == "Label Already Exists" && strings.EqualFold(response.ExistingJobStatus, "RUNNING") {
+		return &streamLoadError{
+			StatusCode: statusCode,
+			Message:    message,
+			Retriable:  false,
+			Ambiguous:  true,
+			Response:   response,
 		}
 	}
 
 	return &streamLoadError{
 		StatusCode: statusCode,
 		Message:    message,
-		Retriable:  retriable,
+		Retriable:  true,
 		Response:   response,
 	}
 }