if (flushInterval != null) {
executor = Executors.newScheduledThreadPool(1);
executor.scheduleWithFixedDelay(this::executeWhenNeeded, 0, flushInterval.millis(), TimeUnit.MILLISECONDS);
} else {
executor = null;
<BUG>}
}</BUG>
public void close() throws InterruptedException {
if (closed) {
return;

if (flushInterval != null) {
executor = Executors.newScheduledThreadPool(1);
executor.scheduleWithFixedDelay(this::executeWhenNeeded, 0, flushInterval.millis(), TimeUnit.MILLISECONDS);
} else {
executor = null;
}
this.pipeline = pipeline;
}
public void close() throws InterruptedException {
if (closed) {
return;
