[Bug] [connector-doris] DorisStreamLoad loading state mismanagement causes RecordBuffer infinite loop during shutdown

[Mrhs121]

Search before asking

• [x] I had searched in the issues and found no similar issues.

What happened

In the exception scenario where the doris fe server returns a status code of 200(for example, an exception like AnalysisException occurs on the fe server)), the following problems will occur

1. In DorisStreamLoad.stopLoad(), the loading state is set to false prematurely, even before the httpClient.execute method returns a response result. This improper timing prevents subsequent error detection mechanisms that rely on the loading state from functioning correctly.
2. This prevents getLoadFailedMsg() from properly collecting error msg form http response
3. The loop in RecordBuffer.stopBufferData() may wait indefinitely due to inability to retrieve error messages https://github.com/apache/seatunnel/blob/40320d866afe51da223dee2da33a5e7c9faecddc/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/RecordBuffer.java#L87-L90
4. This ultimately leads to the task being trapped in a dead loop/infinite loop and causing the task to hang/terminate abnormally

There is also another problem in DorisSinkWriter https://github.com/apache/seatunnel/blob/40320d866afe51da223dee2da33a5e7c9faecddc/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java#L256-L267

1. Incomplete resource cleanup: When flush() throws an IOException, subsequent resource cleanup steps are skipped
2. Thread pool not closed: scheduledExecutorService can not be closed, causing background threads to continue running
3. Connection not released: dorisStreamLoad may not be properly closed, maintaining the connection with Doris

SeaTunnel Version

dev

SeaTunnel Config

Created from the web, without config

Running Command

Created from the web, without command

Error Exception

Doris task blocked by infinite loop and unable to terminate

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

The task cannot be terminated normally

Are you willing to submit PR?

• [x] Yes I am willing to submit a PR!

Code of Conduct

• [x] I agree to follow this project's Code of Conduct

[Mrhs121]

@davidzollo cc, please help review it when you have time, thank you.

[davidzollo]

@davidzollo cc, please help review it when you have time, thank you.

Good job, I'm looking forward to your PR!

[Mrhs121]

@davidzollo Please set the roadmap status to Doing