[Bugfix][Connector-V2] Doris sink check load error before stopLoad to interrupt blocking poll() in RecordBuffer

[Larborator]

Purpose of this pull request

Problem

The stopBufferData() method in RecordBuffer can block indefinitely in a while loop waiting for a buffer from writeQueue.poll(100ms) to send an EOF marker. The checkErrorMessageByStreamLoad() check inside the loop never receives an error signal to interrupt the wait, even when Doris FE has already returned an error response.

flink jstack

Root Cause

1. Buffer pool exhaustion: All buffers are moved from writeQueue to readQueue by the write thread, but the read thread (HTTP upload) doesn't consume them (e.g., due to Doris node restart or network issues). Without consumption, no buffers are recycled back to writeQueue.

2. Error message not propagated: In DorisStreamLoad.stopLoad(), loading = false is set immediately, which prevents getLoadFailedMsg() from checking pendingLoadFuture and setting the error message. The endInput() call blocks before pendingLoadFuture.get() can be reached to parse the error response.

3. Circular dependency: Need EOF to finish → need buffer for EOF → buffer unavailable → need error to interrupt → error not propagated → infinite wait.

Solution

checkDone() before calling stopLoad()

Does this PR introduce any user-facing change?

no

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[zhangshenghang]

thanks @Larborator , I think this PR might solve your problem. Delay the assignment operation of loading=false until after recordStream. endInput(). @Mrhs121 Please confirm it.

• [Fix][Connector-doris] DorisStreamLoad loading state mismanagement causes RecordBuffer infinite loop during shutdown #10060

[Larborator]

thanks @Larborator , I think this PR might solve your problem. Delay the assignment operation of loading=false until after recordStream. endInput(). @Mrhs121 Please confirm it.

• [Fix][Connector-doris] DorisStreamLoad loading state mismanagement causes RecordBuffer infinite loop during shutdown #10060

great work, I'll close this PR.

[Mrhs121]

thanks @Larborator , I think this PR might solve your problem. Delay the assignment operation of loading=false until after recordStream. endInput(). @Mrhs121 Please confirm it.

Confirm that it is the same issue.
FYI, additionally, you need to pay attention to the close method of DorisSinkWriter. If an exception is thrown during the flush process, it may prevent the scheduledExecutorService from shutting down properly, which could cause the SeaTunnel job to fail to terminate.