hotfix: reset bufferCount after buffers are cleared

oceanbase · Mar 13, 2024 · f65041b · f65041b
1 parent 27b0dab
commit f65041b
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...ctor-oceanbase-base/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseWriter.java b/...ctor-oceanbase-base/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseWriter.java
@@ -204,6 +204,7 @@ public synchronized void flush(boolean endOfInput) throws IOException, Interrupt
                         reducedBuffer.clear();
                     }
                 }
+                bufferCount = 0;
 
                 // sync write current record
                 Record record = currentRecord.get();
@@ -221,7 +222,6 @@ public synchronized void flush(boolean endOfInput) throws IOException, Interrupt
                 }
                 metricGroup.getIOMetricGroup().getNumRecordsOutCounter().inc();
                 currentRecord.compareAndSet(record, null);
-                bufferCount = 0;
                 break;
             } catch (Exception e) {
                 LOG.error("OceanBaseWriter flush error, retry times = {}", i, e);