acks: send sequence-0 ack for 0-window batches (#479)

* acks: send sequence-0 ack for 0-window batches

Ensures that at least one ACK is sent in reply to each window, including
those with 0-length, by emitting an empty batch when we receive a 0-window
frame and sending a sequence-0 ACK in reply when we process an empty batch.

This enables upstream users of the protocol to (ab)use a 0-length window
frame and an _ensured_ in-protocol response to validate that we are alive.

Note: 0-sequence ACKs are already used as application-level keep-alives
between when the `ConnectionHandler` receives bytes and when `BeatsHandler`
finishes processing a batch, and therefore should already be handled
gracefully by any existing clients.

* version, changelog

Author: yaauie

.ci/run.sh

@@ -5,6 +5,7 @@ env
 
 set -ex
 
+bundle exec rake test:java
 bundle exec rspec --format=documentation
 
 bundle exec rake test:integration:setup

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 6.7.0
+ - PROTOCOL: adds explicit support for receiving a 0-length window to encapsulate an empty batch. Empty batches are acknowledged with the same 0-sequence ACK's that are used as keep-alives during processing. [#479](https://github.com/logstash-plugins/logstash-input-beats/pull/479)  
+
 ## 6.6.4
  - [DOC] Fix misleading `enrich/source_data` input beats documentation about the Logstash host. [#478](https://github.com/logstash-plugins/logstash-input-beats/pull/478)
 

VERSION

@@ -1 +1 @@
-6.6.4
+6.7.0

lib/tasks/test.rake

@@ -21,6 +21,9 @@ end
 require "logstash/devutils/rake"
 
 namespace :test do
+  task :java do
+    exit(1) unless system './gradlew test'
+  end
   namespace :integration do
     task :setup do
       Rake::Task["test:integration:setup:filebeat"].invoke

src/main/java/org/logstash/beats/BeatsHandler.java

@@ -45,6 +45,10 @@ public void channelRead0(ChannelHandlerContext ctx, Batch batch) {
             logger.debug(format("Received a new payload"));
         }
         try {
+            if (batch.isEmpty()) {
+                logger.debug("Sending 0-seq ACK for empty batch");
+                writeAck(ctx, batch.getProtocol(), 0);
+            }
             for (Message message : batch) {
                 if (logger.isDebugEnabled()) {
                     logger.debug(format("Sending a new message for the listener, sequence: " + message.getSequence()));

src/main/java/org/logstash/beats/BeatsParser.java

@@ -112,6 +112,10 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                     logger.warn("New window size received but the current batch was not complete, sending the current batch");
                     out.add(batch);
                     batchComplete();
+                } else if (batch.getBatchSize() == 0) {
+                    logger.debug("New window size 0 received, sending an empty batch");
+                    out.add(batch);
+                    batchComplete();
                 }
 
                 transition(States.READ_HEADER);

src/test/java/org/logstash/beats/BeatsHandlerTest.java

@@ -121,8 +121,19 @@ public void testAcksLastMessageInBatch() {
         embeddedChannel.writeInbound(batch);
         assertEquals(messageCount, spyListener.getLastMessages().size());
         Ack ack = embeddedChannel.readOutbound();
-        assertEquals(ack.getProtocol(), Protocol.VERSION_1);
-        assertEquals(ack.getSequence(), startSequenceNumber + messageCount - 1);
+        assertEquals(Protocol.VERSION_1, ack.getProtocol());
+        assertEquals(startSequenceNumber + messageCount - 1, ack.getSequence());
+        embeddedChannel.close();
+    }
+
+    @Test
+    public void testAcksZeroSequenceForEmptyBatch() {
+        EmbeddedChannel embeddedChannel = new EmbeddedChannel(new BeatsHandler(spyListener));
+        embeddedChannel.writeInbound(new V2Batch());
+        assertEquals(0, spyListener.getLastMessages().size());
+        Ack ack = embeddedChannel.readOutbound();
+        assertEquals(Protocol.VERSION_2, ack.getProtocol());
+        assertEquals(0, ack.getSequence());
         embeddedChannel.close();
     }
 }

src/test/java/org/logstash/beats/BeatsParserTest.java

@@ -20,6 +20,7 @@
 import static org.hamcrest.Matchers.isA;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 
 public class BeatsParserTest {
@@ -135,6 +136,45 @@ public void testShouldNotCrashOnGarbageData() {
         sendPayloadToParser(randomBufferData);
     }
 
+    @Test
+    public void testV1EmptyWindowEmitsEmptyBatch() {
+        Batch decodedBatch = decodeBatch(new V1Batch());
+
+        assertNotNull(decodedBatch);
+        assertTrue(decodedBatch.isEmpty());
+        assertEquals(0, decodedBatch.getBatchSize());
+        assertEquals(0, decodedBatch.size());
+    }
+    @Test
+    public void testV2EmptyWindowEmitsEmptyBatch() {
+        Batch decodedBatch = decodeBatch(new V2Batch());
+
+        assertNotNull(decodedBatch);
+        assertTrue(decodedBatch.isEmpty());
+        assertEquals(0, decodedBatch.getBatchSize());
+        assertEquals(0, decodedBatch.size());
+    }
+
+
+    @Test
+    public void testV1CompressedFrameEmptyWindowEmitsEmptyBatch() {
+        Batch decodedBatch = decodeCompressedBatch(new V1Batch());
+
+        assertNotNull(decodedBatch);
+        assertTrue(decodedBatch.isEmpty());
+        assertEquals(0, decodedBatch.getBatchSize());
+        assertEquals(0, decodedBatch.size());
+    }
+    @Test
+    public void testV2CompressedFrameEmptyWindowEmitsEmptyBatch() {
+        Batch decodedBatch = decodeCompressedBatch(new V2Batch());
+
+        assertNotNull(decodedBatch);
+        assertTrue(decodedBatch.isEmpty());
+        assertEquals(0, decodedBatch.getBatchSize());
+        assertEquals(0, decodedBatch.size());
+    }
+
 
     @Test
     public void testNegativeJsonPayloadShouldRaiseAnException() throws JsonProcessingException {