Reduce GC pressure by writing directly into the outpustream instead of returning a byte array

Introduce a new (internal) StreamingEncoder interface to be implemented by Encoders that supports writing directly into the output stream instead of returning their results as a byte array.

Update both the AbstractLogstashTcpSocketAppender and the CompositeJsonEncoder to support this new interface. This should hoppefully reduce the amount of short-lived byte arrays created for each log event.

See logfellow#461 for more information.

Author: brenuart

src/main/java/net/logstash/logback/appender/AbstractLogstashTcpSocketAppender.java

@@ -48,6 +48,11 @@
 import net.logstash.logback.appender.destination.PreferPrimaryDestinationConnectionStrategy;
 import net.logstash.logback.appender.listener.TcpAppenderListener;
 import net.logstash.logback.encoder.SeparatorParser;
+import net.logstash.logback.encoder.StreamingEncoder;
+
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.LifecycleAware;
+import com.lmax.disruptor.RingBuffer;
 
 import ch.qos.logback.core.encoder.Encoder;
 import ch.qos.logback.core.joran.spi.DefaultClass;
@@ -59,10 +64,6 @@
 import ch.qos.logback.core.util.CloseUtil;
 import ch.qos.logback.core.util.Duration;
 
-import com.lmax.disruptor.EventHandler;
-import com.lmax.disruptor.LifecycleAware;
-import com.lmax.disruptor.RingBuffer;
-
 /**
  * An {@link AsyncDisruptorAppender} appender that writes
  * events to a TCP {@link Socket} outputStream.
@@ -594,8 +595,9 @@ private void writeEvent(Socket socket, OutputStream outputStream, LogEvent<Event
                  * This is a standard (non-keepAlive) event.
                  * Therefore, we need to send the event.
                  */
-                outputStream.write(encoder.encode(logEvent.event));
-            } else if (hasKeepAliveDurationElapsed(lastSendEndNanoTime, startNanoTime)){
+                encode(logEvent.event, outputStream);
+            } 
+            else if (hasKeepAliveDurationElapsed(lastSendEndNanoTime, startNanoTime)){
                 /*
                  * This is a keep alive event, and the keepAliveDuration has passed,
                  * Therefore, we need to send the keepAliveMessage.
@@ -622,6 +624,21 @@ private void writeEvent(Socket socket, OutputStream outputStream, LogEvent<Event
             }
         }
 
+        
+        @SuppressWarnings("unchecked")
+        private void encode(Event event, OutputStream outputStream) throws IOException {
+            if (encoder instanceof StreamingEncoder) {
+                ((StreamingEncoder<Event>)encoder).encode(event, outputStream);
+            }
+            else {
+                byte[] data = encoder.encode(event);
+                if (data!=null) {
+                    outputStream.write(data);
+                }
+            }
+        }
+        
+        
         private boolean hasKeepAliveDurationElapsed(long lastSentNanoTime, long currentNanoTime) {
             return isKeepAliveEnabled()
                     && lastSentNanoTime + TimeUnit.MILLISECONDS.toNanos(keepAliveDuration.getMilliseconds()) < currentNanoTime;

src/main/java/net/logstash/logback/encoder/CompositeJsonEncoder.java

@@ -15,20 +15,22 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.charset.Charset;
 
 import net.logstash.logback.composite.CompositeJsonFormatter;
 import net.logstash.logback.composite.JsonProviders;
 import net.logstash.logback.decorate.JsonFactoryDecorator;
 import net.logstash.logback.decorate.JsonGeneratorDecorator;
+
 import ch.qos.logback.core.encoder.Encoder;
 import ch.qos.logback.core.encoder.EncoderBase;
 import ch.qos.logback.core.encoder.LayoutWrappingEncoder;
 import ch.qos.logback.core.pattern.PatternLayoutBase;
 import ch.qos.logback.core.spi.DeferredProcessingAware;
 
 public abstract class CompositeJsonEncoder<Event extends DeferredProcessingAware>
-        extends EncoderBase<Event> {
+        extends EncoderBase<Event> implements StreamingEncoder<Event> {
 
     private static final byte[] EMPTY_BYTES = new byte[0];
 
@@ -60,51 +62,45 @@ public CompositeJsonEncoder() {
     protected abstract CompositeJsonFormatter<Event> createFormatter();
 
     @Override
-    public byte[] encode(Event event) {
-        byte[] prefixBytes = doEncodeWrappedToBytes(prefix, event);
-        byte[] suffixBytes = doEncodeWrappedToBytes(suffix, event);
+    public void encode(Event event, OutputStream outputStream) throws IOException {
+        if (!isStarted()) {
+            throw new IllegalStateException("Encoder is not started.");
+        }
 
-        ByteArrayOutputStream outputStream = new ByteArrayOutputStream(
-                minBufferSize
-                + (prefixBytes == null ? 0 : prefixBytes.length)
-                + (suffixBytes == null ? 0 : suffixBytes.length)
-                + lineSeparatorBytes.length);
-        try {
-            if (prefixBytes != null) {
-                outputStream.write(prefixBytes);
-            }   
-            
-            formatter.writeEventToOutputStream(event, outputStream);
-            
-            if (suffixBytes != null) {
-                outputStream.write(suffixBytes);
-            }
-            
-            outputStream.write(lineSeparatorBytes);
-            
+        encode(prefix, event, outputStream);
+        formatter.writeEventToOutputStream(event, outputStream);
+        encode(suffix, event, outputStream);
+        
+        outputStream.write(lineSeparatorBytes);
+    }
+    
+    @Override
+    public byte[] encode(Event event) {
+        try(ByteArrayOutputStream outputStream = new ByteArrayOutputStream(getMinBufferSize())) {
+            encode(event, outputStream);
             return outputStream.toByteArray();
-        } catch (IOException e) {
-            addWarn("Error encountered while encoding log event. "
-                    + "Event: " + event, e);
-            return EMPTY_BYTES;
-        } finally {
-            try {
-                outputStream.close();
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
         }
+        catch (IOException e) {
+            addWarn("Error encountered while encoding log event. Event: " + event, e);
+            return EMPTY_BYTES;
+        } 
     }
 
-    private byte[] doEncodeWrappedToBytes(Encoder<Event> wrapped, Event event) {
-        if (wrapped != null) {
-            return wrapped.encode(event);
+    private void encode(Encoder<Event> encoder, Event event, OutputStream outputStream) throws IOException {
+        if (encoder!=null) {
+            byte[] data = encoder.encode(event);
+            if (data!=null) {
+                outputStream.write(data);
+            }
         }
-        return EMPTY_BYTES;
     }
 
     @Override
     public void start() {
+        if (isStarted()) {
+            return;
+        }
+        
         super.start();
         formatter.setContext(getContext());
         formatter.start();
@@ -151,14 +147,16 @@ private void startWrapped(Encoder<Event> wrapped) {
 
     @Override
     public void stop() {
-        super.stop();
-        formatter.stop();
-        stopWrapped(prefix);
-        stopWrapped(suffix);
+        if (isStarted()) {
+            super.stop();
+            formatter.stop();
+            stopWrapped(prefix);
+            stopWrapped(suffix);
+        }
     }
 
     private void stopWrapped(Encoder<Event> wrapped) {
-        if (wrapped != null && !wrapped.isStarted()) {
+        if (wrapped != null && wrapped.isStarted()) {
             wrapped.stop();
         }
     }

src/main/java/net/logstash/logback/encoder/StreamingEncoder.java

@@ -0,0 +1,27 @@
+/**
+ * Licensed 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 net.logstash.logback.encoder;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Interface implemented by {@link Encoder} that supports writing directly into a {@link OutputStream} 
+ * instead of returning a byte array.
+ */
+public interface StreamingEncoder<Event> {
+
+    public void encode(Event event, OutputStream outputStream) throws IOException;
+
+}