fix retry read buffer problem

spark-doris-connector/src/main/scala/org/apache/doris/spark/writer/DorisWriter.scala

@@ -155,12 +155,10 @@ class DorisWriter(settings: SparkSettings, preCommittedTxnAcc: CollectionAccumul
     override def next(): T = {
       recordCount += 1
       if (batchRetryEnable) {
-        if (isReset && buffer.nonEmpty) {
-          buffer(recordCount)
+        if (isReset) {
+          readBuffer()
         } else {
-          val elem = iterator.next
-          buffer += elem
-          elem
+          writeBufferAndReturn()
         }
       } else {
         iterator.next
@@ -172,8 +170,10 @@ class DorisWriter(settings: SparkSettings, preCommittedTxnAcc: CollectionAccumul
      */
     def reset(): Unit = {
       recordCount = 0
-      isReset = true
-      logger.info("batch iterator is reset")
+      isReset = buffer.nonEmpty
+      if (isReset) {
+        logger.info("buffer is not empty and batch iterator is reset")
+      }
     }
 
     /**
@@ -186,6 +186,20 @@ class DorisWriter(settings: SparkSettings, preCommittedTxnAcc: CollectionAccumul
       }
     }
 
+    private def readBuffer(): T = {
+      if (recordCount == buffer.size) {
+        logger.debug("read buffer end, recordCount:{}, bufferSize: {}", recordCount, buffer.size)
+        isReset = false
+      }
+      buffer(recordCount - 1)
+    }
+
+    private def writeBufferAndReturn(): T = {
+      val elem = iterator.next
+      buffer += elem
+      elem
+    }
+
   }