request commit after a successful flush() (#28)

jerrinot · web-flow · commit af367531b5ce · 2024-09-09T12:34:22.000+02:00
This limits the number of duplicates QuestDB
server has to deal with after error recovery
diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java
@@ -224,6 +224,7 @@ private void flushAndResetCounters() {
             if (sender != null) {
                 sender.flush();
             }
+            context.requestCommit();
             nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;
             pendingRows = 0;
         } catch (LineSenderException | HttpClientException e) {

Original file line number	Diff line number	Diff line change
`@@ -224,6 +224,7 @@ private void flushAndResetCounters() {`
`224`	`224`	`if (sender != null) {`
`225`	`225`	`sender.flush();`
`226`	`226`	`}`
	`227`	`+ context.requestCommit();`
`227`	`228`	`nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;`
`228`	`229`	`pendingRows = 0;`
`229`	`230`	`} catch (LineSenderException \| HttpClientException e) {`