Expose persistent? and checkpoint! on queue write clients for durable input acknowledgement

logstash-core/src/main/java/org/logstash/ackedqueue/Page.java

@@ -215,10 +215,12 @@ public void tailPageCheckpoint() throws IOException {
 
 
     public void ensurePersistedUpto(long seqNum) throws IOException {
+        // lastCheckpointUptoSeqNum is an *exclusive* upper bound: the last checkpoint covers
+        // [minSeqNum, minSeqNum + elementCount - 1], so the first uncovered seqNum is minSeqNum + elementCount.
         long lastCheckpointUptoSeqNum = this.lastCheckpoint.getMinSeqNum() + this.lastCheckpoint.getElementCount();
 
-        // if the last checkpoint for this headpage already included the given seqNum, no need to fsync/checkpoint
-        if (seqNum > lastCheckpointUptoSeqNum) {
+        // checkpoint if seqNum has not been covered yet (>= because lastCheckpointUptoSeqNum is exclusive)
+        if (seqNum >= lastCheckpointUptoSeqNum) {
             // head page checkpoint does a data file fsync
             checkpoint();
         }

logstash-core/src/main/java/org/logstash/ackedqueue/Queue.java

@@ -606,6 +606,26 @@ public void ensurePersistedUpto(long seqNum) throws IOException{
         }
     }
 
+    /**
+     * Guarantee persistence of all elements written to this queue so far.
+     * Fsyncs the head page if it contains writes not yet covered by its last
+     * checkpoint; tail pages are already fsynced at page-rotation time.
+     *
+     * @throws IOException if an IO error occurs
+     */
+    public void ensurePersisted() throws IOException {
+        lock.lock();
+        try {
+            // close() fsyncs everything before nulling headPage; return safely if already closed.
+            if (isClosed()) {
+                return;
+            }
+            this.headPage.ensurePersistedUpto(this.seqNum);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
      * non-blocking queue read
      *

logstash-core/src/main/java/org/logstash/ackedqueue/ext/JRubyAckedQueueExt.java

@@ -146,6 +146,20 @@ public void rubyWrite(ThreadContext context, Event event) {
         }
     }
 
+    public void ensurePersisted(ThreadContext context) {
+        try {
+            this.queue.ensurePersisted();
+        } catch (IOException e) {
+            throw RubyUtil.newRubyIOError(context.runtime, e);
+        }
+    }
+
+    @JRubyMethod(name = "ensure_persisted")
+    public IRubyObject rubyEnsurePersisted(ThreadContext context) {
+        ensurePersisted(context);
+        return context.nil;
+    }
+
     public void write(Event event) {
         try {
             this.queue.write(event);

logstash-core/src/main/java/org/logstash/ext/JRubyAbstractQueueWriteClientExt.java

@@ -55,6 +55,21 @@ public final JRubyAbstractQueueWriteClientExt rubyPushBatch(final ThreadContext
         return this;
     }
 
+    @JRubyMethod(name = "persistent?")
+    public final IRubyObject rubyPersistent(final ThreadContext context) {
+        return context.runtime.newBoolean(isPersistent());
+    }
+
+    @JRubyMethod(name = "checkpoint!")
+    public final JRubyAbstractQueueWriteClientExt rubyCheckpoint(final ThreadContext context) {
+        doCheckpoint(context);
+        return this;
+    }
+
+    protected abstract boolean isPersistent();
+
+    protected abstract void doCheckpoint(ThreadContext context);
+
     protected abstract JRubyAbstractQueueWriteClientExt doPush(ThreadContext context,
                                                                JrubyEventExtLibrary.RubyEvent event) throws InterruptedException;
 

logstash-core/src/main/java/org/logstash/ext/JRubyWrappedWriteClientExt.java

@@ -127,6 +127,16 @@ public IRubyObject pushBatch(final ThreadContext context,
         ));
     }
 
+    @JRubyMethod(name = "persistent?")
+    public IRubyObject rubyPersistent(final ThreadContext context) {
+        return writeClient.rubyPersistent(context);
+    }
+
+    @JRubyMethod(name = "checkpoint!")
+    public IRubyObject rubyCheckpoint(final ThreadContext context) {
+        return writeClient.rubyCheckpoint(context);
+    }
+
     /**
      * @param context Ruby {@link ThreadContext}
      * @return Empty {@link RubyArray}

logstash-core/src/main/java/org/logstash/ext/JrubyAckedWriteClientExt.java

@@ -54,6 +54,16 @@ private JrubyAckedWriteClientExt(final Ruby runtime, final RubyClass metaClass,
         this.queue = queue;
     }
 
+    @Override
+    protected boolean isPersistent() {
+        return true;
+    }
+
+    @Override
+    protected void doCheckpoint(final ThreadContext context) {
+        queue.ensurePersisted(context);
+    }
+
     @Override
     protected JRubyAbstractQueueWriteClientExt doPush(final ThreadContext context,
                                                       final JrubyEventExtLibrary.RubyEvent event) {

logstash-core/src/main/java/org/logstash/ext/JrubyMemoryWriteClientExt.java

@@ -55,6 +55,16 @@ public static JrubyMemoryWriteClientExt create(
                 RubyUtil.MEMORY_WRITE_CLIENT_CLASS, queue);
     }
 
+    @Override
+    protected boolean isPersistent() {
+        return false;
+    }
+
+    @Override
+    protected void doCheckpoint(final ThreadContext context) {
+        throw context.runtime.newNotImplementedError("checkpoint! called on non-persistent queue");
+    }
+
     @Override
     protected JRubyAbstractQueueWriteClientExt doPush(final ThreadContext context,
                                                       final JrubyEventExtLibrary.RubyEvent event) throws InterruptedException {

logstash-core/src/test/java/org/logstash/ackedqueue/QueueTest.java

@@ -1194,4 +1194,45 @@ public void writeToClosedQueueException() throws Exception {
         });
         assertThat(qre.getMessage(), containsString("Tried to write to a closed queue."));
     }
+
+    @Test
+    public void ensurePersistedFsyncsHeadPageUpToLastWrite() throws IOException {
+        // Use a high checkpointMaxWrites so writes do NOT auto-checkpoint (TestSettings default is 1)
+        Settings settings = SettingsImpl.builder(
+            TestSettings.persistedQueueSettings(1024, dataPath)
+        ).checkpointMaxWrites(1024).build();
+        try (Queue q = new Queue(settings)) {
+            q.open();
+            q.write(new StringElement("foo"));
+            q.write(new StringElement("bar"));
+
+            // head checkpoint does not yet cover the two writes
+            Checkpoint before = q.getCheckpointIO().read("checkpoint.head");
+            assertThat(before.getElementCount(), is(0));
+
+            q.ensurePersisted();
+
+            Checkpoint after = q.getCheckpointIO().read("checkpoint.head");
+            assertThat(after.getElementCount(), is(2));
+        }
+    }
+
+    @Test
+    public void ensurePersistedIsNoOpWhenNothingNewWasWritten() throws IOException {
+        // Use a high checkpointMaxWrites so writes do NOT auto-checkpoint (TestSettings default is 1)
+        Settings settings = SettingsImpl.builder(
+            TestSettings.persistedQueueSettings(1024, dataPath)
+        ).checkpointMaxWrites(1024).build();
+        try (Queue q = new Queue(settings)) {
+            q.open();
+            q.ensurePersisted(); // empty queue: must not throw
+
+            q.write(new StringElement("foo"));
+            q.ensurePersisted();
+            Checkpoint first = q.getCheckpointIO().read("checkpoint.head");
+            q.ensurePersisted(); // second call: nothing new, must not re-checkpoint or throw
+            Checkpoint second = q.getCheckpointIO().read("checkpoint.head");
+            assertThat(second.getElementCount(), is(first.getElementCount()));
+        }
+    }
 }

logstash-core/src/test/java/org/logstash/ext/JRubyWrappedWriteClientExtTest.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.logstash.ext;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import org.jruby.exceptions.NotImplementedError;
+import org.jruby.runtime.ThreadContext;
+import org.junit.Test;
+import org.logstash.RubyTestBase;
+import org.logstash.RubyUtil;
+import org.logstash.instrument.metrics.MockNamespacedMetric;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+
+/**
+ * Tests for {@link JRubyWrappedWriteClientExt}.
+ */
+public final class JRubyWrappedWriteClientExtTest extends RubyTestBase {
+
+    private JRubyWrappedWriteClientExt wrappedClient() {
+        final JrubyMemoryWriteClientExt delegate =
+            JrubyMemoryWriteClientExt.create(new ArrayBlockingQueue<>(10));
+        final JRubyWrappedWriteClientExt wrapped = new JRubyWrappedWriteClientExt(
+            RubyUtil.RUBY, RubyUtil.WRAPPED_WRITE_CLIENT_CLASS);
+        return wrapped.initialize(delegate, "test-pipeline",
+            MockNamespacedMetric.create(), RubyUtil.RUBY.newString("test-plugin"));
+    }
+
+    @Test
+    public void delegatesPersistentToWrappedClient() {
+        final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+        assertFalse(wrappedClient().rubyPersistent(context).isTrue());
+    }
+
+    @Test
+    public void delegatesCheckpointToWrappedClient() {
+        final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+        assertThrows(NotImplementedError.class,
+            () -> wrappedClient().rubyCheckpoint(context));
+    }
+}

logstash-core/src/test/java/org/logstash/ext/JrubyAckedWriteClientExtTest.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.logstash.ext;
+
+import java.io.IOException;
+import org.jruby.runtime.ThreadContext;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.logstash.Event;
+import org.logstash.RubyTestBase;
+import org.logstash.RubyUtil;
+import org.logstash.ackedqueue.Checkpoint;
+import org.logstash.ackedqueue.SettingsImpl;
+import org.logstash.ackedqueue.ext.JRubyAckedQueueExt;
+import org.logstash.plugins.NamespacedMetricImpl;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link JrubyAckedWriteClientExt}.
+ */
+public final class JrubyAckedWriteClientExtTest extends RubyTestBase {
+
+    @Rule
+    public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+    @Test
+    public void reportsPersistentAndCheckpointFsyncsQueue() throws IOException {
+        final String dataPath = temporaryFolder.newFolder("data").getPath();
+        final JRubyAckedQueueExt queue = JRubyAckedQueueExt.create(
+            SettingsImpl.fileSettingsBuilder(dataPath)
+                .elementClass(Event.class)
+                .capacity(1024 * 1024)
+                .maxUnread(0)
+                .queueMaxBytes(0)
+                .checkpointMaxAcks(1024)
+                .checkpointMaxWrites(1024)
+                .build(),
+            NamespacedMetricImpl.getNullMetric());
+        queue.open();
+        try {
+            final JrubyAckedWriteClientExt client = JrubyAckedWriteClientExt.create(queue);
+            final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+
+            assertTrue(client.rubyPersistent(context).isTrue());
+
+            queue.rubyWrite(context, new Event());
+            client.rubyCheckpoint(context);
+
+            final Checkpoint head = queue.getQueue().getCheckpointIO().read("checkpoint.head");
+            assertThat(head.getElementCount(), is(1));
+        } finally {
+            queue.close();
+        }
+    }
+}

logstash-core/src/test/java/org/logstash/ext/JrubyMemoryWriteClientExtTest.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.logstash.ext;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import org.jruby.exceptions.NotImplementedError;
+import org.jruby.runtime.ThreadContext;
+import org.junit.Test;
+import org.logstash.RubyTestBase;
+import org.logstash.RubyUtil;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+
+/**
+ * Tests for {@link JrubyMemoryWriteClientExt}.
+ */
+public final class JrubyMemoryWriteClientExtTest extends RubyTestBase {
+
+    @Test
+    public void reportsNotPersistent() {
+        final JrubyMemoryWriteClientExt client =
+            JrubyMemoryWriteClientExt.create(new ArrayBlockingQueue<>(10));
+        final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+        assertFalse(client.rubyPersistent(context).isTrue());
+    }
+
+    @Test
+    public void checkpointRaisesNotImplemented() {
+        final JrubyMemoryWriteClientExt client =
+            JrubyMemoryWriteClientExt.create(new ArrayBlockingQueue<>(10));
+        final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
+        assertThrows(NotImplementedError.class, () -> client.rubyCheckpoint(context));
+    }
+}