[Improve](go) Align stream load retry policy with the Flink connector

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.

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,46 +551,34 @@ 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
 		}
 
 		if retryDeadline.IsZero() {
 			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
 		}
 
 		backoff := c.retryBackoffDelay(attempts)
 		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)
 	}
 }