Skip commit in CloseOnFlushWriterWrapper if a commit has already been invoked on the underlying writer.[]
Closes from sv2000/closeOnFlushWriterWrapper
@@ -57,6 +57,8 @@ public class CloseOnFlushWriterWrapper<D> extends WriterWrapper<D> implements De
private DataWriter<D> writer;
private final Supplier<DataWriter<D>> writerSupplier;
private boolean closed;
+ private boolean committed;
+
// is the close functionality enabled?
private final boolean closeOnFlush;
private final ControlMessageHandler controlMessageHandler;
@@ -90,6 +92,7 @@ public class CloseOnFlushWriterWrapper<D> extends WriterWrapper<D> implements De
if (this.closed) {
this.writer = writerSupplier.get();
this.closed = false;
+ this.committed = false;
}
this.writer.writeEnvelope(record);
}
@@ -104,13 +107,15 @@ public class CloseOnFlushWriterWrapper<D> extends WriterWrapper<D> implements De
@Override
public void commit() throws IOException {
+ if (!this.committed) {
writer.commit();
+ this.committed = true;
+ }
}
@Override
public void cleanup() throws IOException {
writer.cleanup();
-
}
@Override
