Add size size key in option at out_forward

Sending array's length with msgpack is always 3, so remove FORWARD_HEADER_EXT

lib/fluent/compat/output.rb

@@ -111,6 +111,10 @@ module ChunkSizeCompatMixin
       def size
         self.bytesize
       end
+
+      def contains_size
+        @size + @adding_size
+      end
     end
 
     module BufferedChunkMixin

lib/fluent/plugin/out_forward.rb

@@ -276,16 +276,10 @@ def rebuild_weight_array
       @weight_array = weight_array
     end
 
-    # MessagePack FixArray length = 3 (if @extend_internal_protocol)
-    #                             = 2 (else)
-    FORWARD_HEADER = [0x92].pack('C').freeze
-    FORWARD_HEADER_EXT = [0x93].pack('C').freeze
+    # MessagePack FixArray length is 3
+    FORWARD_HEADER = [0x93].pack('C').freeze
     def forward_header
-      if @extend_internal_protocol
-        FORWARD_HEADER_EXT
-      else
-        FORWARD_HEADER
-      end
+      FORWARD_HEADER
     end
 
     #FORWARD_TCP_HEARTBEAT_DATA = FORWARD_HEADER + ''.to_msgpack + [].to_msgpack
@@ -336,8 +330,9 @@ def send_data(node, tag, chunk)
         # writeRawBody(packed_es)
         chunk.write_to(sock)
 
+        option = { 'size' => chunk.contains_size }
+
         if @extend_internal_protocol
-          option = {}
           option['chunk'] = Base64.encode64(chunk.unique_id) if @require_ack_response
           sock.write option.to_msgpack
 
@@ -374,6 +369,8 @@ def send_data(node, tag, chunk)
               raise ForwardOutputACKTimeoutError, "node #{node.host}:#{node.port} does not return ACK"
             end
           end
+        else
+          sock.write option.to_msgpack
         end
 
         node.heartbeat(false)

test/plugin/test_out_forward.rb

@@ -120,6 +120,70 @@ def test_wait_response_timeout_config
     assert_equal 2, d.instance.ack_response_timeout
   end
 
+  def test_sending_contains_with_ack
+    target_input_driver = create_target_input_driver(true)
+
+    d = create_driver(CONFIG + %[
+      ack_response_timeout 1s
+    ])
+
+    time = Time.parse("2011-01-02 13:14:15 UTC").to_i
+
+    records = [
+      {"a" => 1},
+      {"a" => 2}
+    ]
+    d.register_run_post_condition do
+      d.instance.responses.length == 1
+    end
+
+    target_input_driver.run do
+      d.run do
+        records.each do |record|
+          d.emit record, time
+        end
+      end
+    end
+
+    emits = target_input_driver.emits
+    assert_equal ['test', time, records[0]], emits[0]
+    assert_equal ['test', time, records[1]], emits[1]
+
+    assert_equal target_input_driver.instance.storage[0]['size'], 2
+  end
+
+  def test_sending_contains_without_ack
+    target_input_driver = create_target_input_driver(true)
+
+    d = create_driver(CONFIG + %[
+      ack_response_timeout 1s
+    ])
+
+    time = Time.parse("2011-01-02 13:14:15 UTC").to_i
+
+    records = [
+      {"a" => 1},
+      {"a" => 2}
+    ]
+    d.register_run_post_condition do
+      d.instance.responses.length == 1
+    end
+
+    target_input_driver.run do
+      d.run do
+        records.each do |record|
+          d.emit record, time
+        end
+      end
+    end
+
+    emits = target_input_driver.emits
+    assert_equal ['test', time, records[0]], emits[0]
+    assert_equal ['test', time, records[1]], emits[1]
+
+    assert_equal target_input_driver.instance.storage[0]['size'], 2
+  end
+
   def test_send_with_time_as_integer
     target_input_driver = create_target_input_driver
 
@@ -379,12 +443,18 @@ def create_target_input_driver(do_respond=false, disconnect=false, conf=TARGET_C
     DummyEngineDriver.new(Fluent::ForwardInput) {
       handler_class = Class.new(Fluent::ForwardInput::Handler) { |klass|
         attr_reader :chunk_counter # for checking if received data is successfully deserialized
+        attr_reader :storage
 
         def initialize(sock, log, on_message)
           @sock = sock
           @log = log
           @chunk_counter = 0
-          @on_message = on_message
+          @storage = []
+          @on_message = ->(msg, chunk_size, source) {
+            option = on_message.call(msg, chunk_size, source)
+            @storage << option
+            option
+          }
           @source = nil
         end
 
@@ -409,7 +479,7 @@ def close
         end
       }
 
-      define_method(:start) do
+      define_method(:_start) do
         @thread = Thread.new do
           Socket.tcp_server_loop(@bind, @port) do |sock, client_addrinfo|
             begin
@@ -431,12 +501,15 @@ def close
                 sock.close_write
                 sock.close
               end
+              @storage = handler.storage
             end
           end
         end
       end
 
-      def shutdown
+      attr_reader :storage
+
+      def _shutdown
         @thread.kill
         @thread.join
       end
@@ -457,9 +530,24 @@ def test_heartbeat_type_none
     assert_equal node.available, true
   end
 
+  module SuppressCallSuperMixin
+    def start
+      _start
+    end
+
+    def before_shutdown
+      # nothing
+    end
+
+    def shutdown
+      _shutdown
+    end
+  end
+
   class DummyEngineDriver < Fluent::Test::TestDriver
     def initialize(klass, &block)
       super(klass, &block)
+      @instance.class.prepend(SuppressCallSuperMixin)
       @engine = DummyEngineClass.new
     end