diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb
index fe9c253ff0..81468c3979 100644
--- a/lib/fluent/plugin/out_forward.rb
+++ b/lib/fluent/plugin/out_forward.rb
@@ -525,17 +525,28 @@ def ack_reader
 
     def ack_check(select_interval)
       @ack_handler.collect_response(select_interval) do |chunk_id, node, sock, result|
-        @connection_manager.close(sock)
-
         case result
         when AckHandler::Result::SUCCESS
+          @connection_manager.close(sock)
           commit_write(chunk_id)
         when AckHandler::Result::FAILED
+          # The ack timed out or the peer closed the connection. A keepalive
+          # socket may still have an in-flight ack for this chunk, so drop it
+          # instead of returning it to the reuse pool; otherwise a late ack
+          # could later be read against a chunk sent on the reused socket (#4057).
+          @connection_manager.discard(sock)
           node&.disable!
           rollback_write(chunk_id, update_retry: false) if chunk_id
         when AckHandler::Result::CHUNKID_UNMATCHED
+          # The ack read from this socket belongs to a different chunk than the
+          # one tracked for it: the keepalive connection's ack stream is out of
+          # sync (e.g. a stale ack left over from before an aggregator restart).
+          # Discard the socket so the following acks are not each offset by one,
+          # which would otherwise flood the log with mismatch warnings (#4057).
+          @connection_manager.discard(sock)
           rollback_write(chunk_id, update_retry: false)
         else
+          @connection_manager.close(sock)
           log.warn("BUG: invalid status #{result} #{chunk_id}")
 
           if chunk_id
diff --git a/lib/fluent/plugin/out_forward/connection_manager.rb b/lib/fluent/plugin/out_forward/connection_manager.rb
index 3b8ad32595..75cfe539f1 100644
--- a/lib/fluent/plugin/out_forward/connection_manager.rb
+++ b/lib/fluent/plugin/out_forward/connection_manager.rb
@@ -78,6 +78,24 @@ def close(sock)
         end
       end
 
+      # Drop a socket instead of returning it to the keepalive reuse pool.
+      #
+      # Used when a chunk's ack timed out or came back with a mismatched chunk
+      # id: such a connection may still carry an in-flight/stale ack for a
+      # previous chunk. Reusing it would let that stale ack be read against a
+      # chunk sent on the reused socket, permanently offsetting every following
+      # ack and flooding the log with 'ack in response and chunk id in sent data
+      # are different' warnings (see #4057). Revoking removes it from the pool so
+      # the next chunk uses a fresh connection with a clean ack stream.
+      def discard(sock)
+        if @socket_cache
+          @socket_cache.revoke(sock)
+        else
+          sock.close_write rescue nil
+          sock.close rescue nil
+        end
+      end
+
       private
 
       def connect_keepalive(host:, port:, hostname:, ack: nil)
diff --git a/test/plugin/out_forward/test_connection_manager.rb b/test/plugin/out_forward/test_connection_manager.rb
index 5c528cc768..a4133d8746 100644
--- a/test/plugin/out_forward/test_connection_manager.rb
+++ b/test/plugin/out_forward/test_connection_manager.rb
@@ -142,4 +142,66 @@ class ConnectionManager < Test::Unit::TestCase
       end
     end
   end
+
+  sub_test_case '#discard' do
+    test 'closes the socket directly when socket_cache is nil' do
+      sock = 'sock'
+      cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new(
+        log: $log,
+        secure: false,
+        connection_factory: -> (_, _, _) { sock },
+        socket_cache: nil,
+      )
+
+      mock(sock).close_write.once
+      mock(sock).close.once
+      cm.discard(sock)
+    end
+
+    test 'revokes a keepalive socket instead of returning it to the reuse pool' do
+      cache = Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log)
+      mock(cache).revoke('sock').once
+      mock(cache).checkin('sock').never
+
+      cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new(
+        log: $log,
+        secure: false,
+        connection_factory: -> (_, _, _) { 'sock' },
+        socket_cache: cache,
+      )
+
+      cm.discard('sock')
+    end
+
+    test 'a discarded keepalive socket is not reused, unlike a checked-in one' do
+      ack = Object.new
+      def ack.enqueue(_sock); end
+
+      # Baseline: a checked-in socket is returned to the pool and reused.
+      reused = []
+      reuse_cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new(
+        log: $log,
+        secure: false,
+        connection_factory: -> (_, _, _) { s = "sock#{reused.size}"; reused << s; s },
+        socket_cache: Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log),
+      )
+      reuse_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| }
+      reuse_cm.close('sock0')
+      reuse_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| }
+      assert_equal(['sock0'], reused)
+
+      # A discarded socket is dropped, so the next checkout builds a fresh one.
+      discarded = []
+      discard_cm = Fluent::Plugin::ForwardOutput::ConnectionManager.new(
+        log: $log,
+        secure: false,
+        connection_factory: -> (_, _, _) { s = "sock#{discarded.size}"; discarded << s; s },
+        socket_cache: Fluent::Plugin::ForwardOutput::SocketCache.new(10, $log),
+      )
+      discard_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| }
+      discard_cm.discard('sock0')
+      discard_cm.connect(host: 'host', port: 1234, hostname: 'name', ack: ack) { |_s, _ri| }
+      assert_equal(['sock0', 'sock1'], discarded)
+    end
+  end
 end
diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index 0c44e1d74e..1f4821df4e 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -1429,4 +1429,70 @@ def plugin_id_for_test?
     logs = d.logs
     assert{ logs.any?{|log| log.include?('handshake timeout after 1.0s') } }
   end
+
+  sub_test_case 'ack_check discards tainted keepalive sockets (#4057)' do
+    def create_ack_driver
+      create_driver(config + %[
+        require_ack_response true
+        keepalive true
+      ])
+    end
+
+    # A keepalive socket whose ack failed or came back mismatched may still
+    # carry an in-flight/stale ack for a previous chunk. Returning it to the
+    # reuse pool lets that stale ack be read against a chunk sent on the reused
+    # socket, permanently offsetting every following ack and flooding the log
+    # with 'ack in response and chunk id in sent data are different' warnings.
+    data(
+      'mismatched ack' => Fluent::Plugin::ForwardOutput::AckHandler::Result::CHUNKID_UNMATCHED,
+      'failed ack' => Fluent::Plugin::ForwardOutput::AckHandler::Result::FAILED,
+    )
+    test 'discards the socket instead of reusing it' do |result|
+      @d = d = create_ack_driver
+      instance = d.instance
+      node = instance.nodes.first
+      sock = 'sock'
+
+      # One-shot: only the explicit ack_check below yields; the collect_response
+      # that after_shutdown triggers during teardown stays silent.
+      fake_ack_handler = Object.new
+      fake_ack_handler.instance_variable_set(:@yielded, false)
+      fake_ack_handler.define_singleton_method(:collect_response) do |_interval, &block|
+        next if @yielded
+        @yielded = true
+        block.call('chunk-id', node, sock, result)
+      end
+      instance.instance_variable_set(:@ack_handler, fake_ack_handler)
+      stub(instance).rollback_write
+
+      cm = instance.instance_variable_get(:@connection_manager)
+      mock(cm).discard(sock).once
+      mock(cm).close(sock).never
+
+      instance.__send__(:ack_check, 0)
+    end
+
+    test 'reuses the socket on a successful ack' do
+      @d = d = create_ack_driver
+      instance = d.instance
+      node = instance.nodes.first
+      sock = 'sock'
+
+      fake_ack_handler = Object.new
+      fake_ack_handler.instance_variable_set(:@yielded, false)
+      fake_ack_handler.define_singleton_method(:collect_response) do |_interval, &block|
+        next if @yielded
+        @yielded = true
+        block.call('chunk-id', node, sock, Fluent::Plugin::ForwardOutput::AckHandler::Result::SUCCESS)
+      end
+      instance.instance_variable_set(:@ack_handler, fake_ack_handler)
+      stub(instance).commit_write
+
+      cm = instance.instance_variable_get(:@connection_manager)
+      mock(cm).close(sock).once
+      mock(cm).discard(sock).never
+
+      instance.__send__(:ack_check, 0)
+    end
+  end
 end