[Bug] connector can't recover after doris become normal

[kele1997]

Search before asking

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

Version

https://github.com/apache/doris-flink-connector/tree/release-25.1.0

What's Wrong?

doris-flink-connector/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriter.java

Line 60 in 65d8c00

private transient volatile Exception flushException = null;

DorisBatchWriter has a field named flushException.
When doris is down, intervalFlush() will set flushException, but after doris is up, no one will clear flushException, and every time connector write data, it will check the flushException, so connector will stuck, and never recover from last exception.

doris-flink-connector/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriter.java

Lines 116 to 123 in 65d8c00

	private void intervalFlush() {
	try {
	boolean flush = batchStreamLoad.intervalFlush();
	LOG.debug("interval flush trigger, flush: {}", flush);
	} catch (Exception e) {
	flushException = e;
	}
	}

doris-flink-connector/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchWriter.java

Lines 175 to 180 in 65d8c00

	private void checkFlushException() {
	if (flushException != null) {
	throw new RuntimeException("Writing records to streamload failed.", flushException);
	}
	}
	}

What You Expected?

when doris is up, the connector should recover from last exception

How to Reproduce?

No response

Anything Else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[JNSimba]

When Flink restarts internally, should this be reinitialized?

[xxntti3n]

In my case , When MySQL Flink CDC is in the snapshot phase, and Doris restarts causing the Flink job to restart as well, the Flink job stops receiving any further data changes after the restart. However, if the job is in the streaming phase, a Flink job restart or reinitialization still allows it to continue capturing and processing new record changes.