diff --git a/lib/dalli/protocol/base.rb b/lib/dalli/protocol/base.rb
index 8bbeaa95..c04cded0 100644
--- a/lib/dalli/protocol/base.rb
+++ b/lib/dalli/protocol/base.rb
@@ -96,6 +96,7 @@ def unlock!; end
       # Returns nothing.
       def pipeline_response_setup
         verify_pipelined_state(:getkq)
+        @pipeline_error = nil
         write_noop
         response_buffer.reset
       end
@@ -112,20 +113,20 @@ def pipeline_next_responses
 
         response_buffer.read
 
-        status, cas, key, value = response_buffer.process_single_getk_response
         # status is not nil only if we have a full response to parse
         # in the buffer
-        until status.nil?
-          # If the status is ok and key is nil, then this is the response
-          # to the noop at the end of the pipeline
-          finish_pipeline && break if status && key.nil?
+        loop do
+          response = response_buffer.process_single_getk_response
+          break if response.first.nil?
 
-          # If the status is ok and the key is not nil, then this is a
-          # getkq response with a value that we want to set in the response hash
-          values[key] = [value, cas] unless key.nil?
+          process_pipeline_getk_response(values, response)
+          break if pipeline_complete?
+        end
 
-          # Get the next response from the buffer
-          status, cas, key, value = response_buffer.process_single_getk_response
+        if pipeline_complete? && @pipeline_error
+          error = @pipeline_error
+          @pipeline_error = nil
+          raise error
         end
 
         values
@@ -138,7 +139,23 @@ def pipeline_next_responses
       # disconnected, and the exception is swallowed.
       #
       # Returns nothing.
+      def process_pipeline_getk_response(values, response)
+        status, cas, key, value, error = response
+        @pipeline_error ||= error if error
+
+        # If the status is ok and key is nil, then this is the response
+        # to the noop at the end of the pipeline
+        if status && key.nil?
+          finish_pipeline
+        elsif !key.nil?
+          # If the status is ok and the key is not nil, then this is a
+          # getkq response with a value that we want to set in the response hash
+          values[key] = [value, cas]
+        end
+      end
+
       def pipeline_abort
+        @pipeline_error = nil
         response_buffer.clear
         @connection_manager.abort_request!
         return true unless connected?
diff --git a/lib/dalli/protocol/meta.rb b/lib/dalli/protocol/meta.rb
index 92fefe7c..b3b25ea7 100644
--- a/lib/dalli/protocol/meta.rb
+++ b/lib/dalli/protocol/meta.rb
@@ -82,12 +82,18 @@ def read_multi_req(keys, req_options = nil)
           @connection_manager.flush
 
           terminator_length = TERMINATOR.length
+          error = nil
           while (line = @connection_manager.readline)
             break if line == TERMINATOR || line[0, 2] == 'MN'
+
+            tokens = line.split
+            if response_processor.error_response?(tokens.first)
+              error ||= response_processor.response_error_from_line(line.chomp(TERMINATOR))
+              next
+            end
             next unless line[0, 3] == 'VA '
 
             # VA value_length flags key
-            tokens = line.split
             value = @connection_manager.read_exact(tokens[1].to_i)
             bitflags = optimized_for_raw ? 0 : @response_processor.bitflags_from_tokens(tokens)
             @connection_manager.read_exact(terminator_length) # read the terminator
@@ -103,6 +109,8 @@ def read_multi_req(keys, req_options = nil)
             attributes['hit_count'] = results.size
             attributes['miss_count'] = keys.size - results.size
           end
+
+          raise error if error
         end
 
         results
@@ -131,11 +139,17 @@ def read_multi_with_status_req(keys, req_options = nil)
           @connection_manager.flush
 
           terminator_length = TERMINATOR.length
+          error = nil
           while (line = @connection_manager.readline)
             break if line == TERMINATOR || line[0, 2] == 'MN'
-            next unless line[0, 3] == 'VA '
 
             tokens = line.split
+            if response_processor.error_response?(tokens.first)
+              error ||= response_processor.response_error_from_line(line.chomp(TERMINATOR))
+              next
+            end
+            next unless line[0, 3] == 'VA '
+
             value = @connection_manager.read_exact(tokens[1].to_i)
             bitflags = optimized_for_raw ? 0 : response_processor.bitflags_from_tokens(tokens)
             @connection_manager.read_exact(terminator_length)
@@ -167,6 +181,8 @@ def read_multi_with_status_req(keys, req_options = nil)
             attributes['miss_count'] = keys.size - fresh_hit_count
             attributes['stale_count'] = stale_count
           end
+
+          raise error if error
         end
 
         results
diff --git a/lib/dalli/protocol/meta/response_processor.rb b/lib/dalli/protocol/meta/response_processor.rb
index e3a763be..4144eb06 100644
--- a/lib/dalli/protocol/meta/response_processor.rb
+++ b/lib/dalli/protocol/meta/response_processor.rb
@@ -21,7 +21,10 @@ class ResponseProcessor
         STAT = 'STAT'
         VA = 'VA'
         VERSION = 'VERSION'
+        CLIENT_ERROR = 'CLIENT_ERROR'
+        ERROR = 'ERROR'
         SERVER_ERROR = 'SERVER_ERROR'
+        ERROR_RESPONSES = [CLIENT_ERROR, SERVER_ERROR, ERROR].freeze
 
         def initialize(io_source, value_marshaller)
           @io_source = io_source
@@ -141,25 +144,43 @@ def version
         end
 
         def consume_all_responses_until_mn
-          tokens = next_line_to_tokens
+          error = nil
+
+          each_response_until_mn do |line, tokens|
+            error ||= response_error_from_line(line) if error_response?(tokens.first)
+          end
+
+          raise error if error
 
-          tokens = next_line_to_tokens while tokens.first != MN
           true
         end
 
         # In quiet mode, only error responses (NF) are sent, success (HD) is suppressed.
         # Returns the count of NF (not found) responses.
         def count_not_found_responses_until_mn
+          error = nil
           not_found_count = 0
-          tokens = next_line_to_tokens
 
-          while tokens.first != MN
+          each_response_until_mn do |line, tokens|
+            error ||= response_error_from_line(line) if error_response?(tokens.first)
             not_found_count += 1 if tokens.first == NF
-            tokens = next_line_to_tokens
           end
+
+          raise error if error
+
           not_found_count
         end
 
+        def each_response_until_mn
+          loop do
+            line = read_line
+            tokens = line&.split || []
+            break if tokens.first == MN
+
+            yield line, tokens
+          end
+        end
+
         def tokens_from_header_buffer(buf)
           header = header_from_buffer(buf)
           tokens = header.split
@@ -170,7 +191,7 @@ def tokens_from_header_buffer(buf)
 
         def full_response_from_buffer(tokens, body, resp_size)
           value = @value_marshaller.retrieve(body, bitflags_from_tokens(tokens))
-          [resp_size, tokens.first == VA, cas_from_tokens(tokens), key_from_tokens(tokens), value]
+          [resp_size, tokens.first == VA, cas_from_tokens(tokens), key_from_tokens(tokens), value, nil]
         end
 
         ##
@@ -185,20 +206,26 @@ def full_response_from_buffer(tokens, body, resp_size)
         ##
         def getk_response_from_buffer(buf)
           # There's no header in the buffer, so don't advance
-          return [0, nil, nil, nil, nil] unless contains_header?(buf)
+          return [0, nil, nil, nil, nil, nil] unless contains_header?(buf)
 
           tokens, header_len, body_len = tokens_from_header_buffer(buf)
 
-          # We have a complete response that has no body.
-          # This is either the response to the terminating
-          # noop or, if the status is not MN, an intermediate
-          # error response that needs to be discarded.
-          return [header_len, true, nil, nil, nil] if body_len.zero?
+          # Error responses from an upstream proxy/partition are explicitly
+          # discarded and must not be returned to callers as raw values or
+          # associated with any requested key.
+          if error_response?(tokens.first)
+            return [header_len, false, nil, nil, nil, response_error_from_line(header_from_buffer(buf))]
+          end
+
+          # We have a complete response that has no body.  Only MN is the
+          # response to the terminating noop; other body-less responses are
+          # discarded.
+          return [header_len, tokens.first == MN, nil, nil, nil, nil] if body_len.zero?
 
           resp_size = header_len + body_len + TERMINATOR.length
           # The header is in the buffer, but the body is not.  As we don't have
           # a complete response, don't advance the buffer
-          return [0, nil, nil, nil, nil] unless buf.bytesize >= resp_size
+          return [0, nil, nil, nil, nil, nil] unless buf.bytesize >= resp_size
 
           # The full response is in our buffer, so parse it and return
           # the values
@@ -220,11 +247,22 @@ def error_on_unexpected!(expected_codes)
 
           return tokens if expected_codes.include?(tokens.first)
 
-          raise Dalli::ServerError, line if tokens.first == SERVER_ERROR
+          raise response_error_from_line(line) if error_response?(tokens.first)
 
           raise Dalli::DalliError, "Response error: #{line}"
         end
 
+        def response_error_from_line(line)
+          tokens = line&.split || []
+          return Dalli::ServerError.new(line) if tokens.first == SERVER_ERROR
+
+          Dalli::DalliError.new("Response error: #{line}")
+        end
+
+        def error_response?(code)
+          ERROR_RESPONSES.include?(code)
+        end
+
         def meta_flags_from_tokens(tokens)
           {
             c: cas_from_tokens(tokens),
diff --git a/lib/dalli/protocol/response_buffer.rb b/lib/dalli/protocol/response_buffer.rb
index 6fcd6482..67fa0111 100644
--- a/lib/dalli/protocol/response_buffer.rb
+++ b/lib/dalli/protocol/response_buffer.rb
@@ -22,9 +22,9 @@ def read
       # Attempts to process a single response from the buffer.  Starts
       # by advancing the buffer to the specified start position
       def process_single_getk_response
-        bytes, status, cas, key, value = @response_processor.getk_response_from_buffer(@buffer)
+        bytes, status, cas, key, value, error = @response_processor.getk_response_from_buffer(@buffer)
         advance(bytes)
-        [status, cas, key, value]
+        [status, cas, key, value, error]
       end
 
       # Advances the internal response buffer by bytes_to_advance
diff --git a/test/protocol/meta/test_response_processor.rb b/test/protocol/meta/test_response_processor.rb
index 707e1be1..79192059 100644
--- a/test/protocol/meta/test_response_processor.rb
+++ b/test/protocol/meta/test_response_processor.rb
@@ -12,6 +12,43 @@ def read_line
   end
 end
 
+class PipelinedResponseErrorTestBuffer
+  attr_reader :processed_count
+
+  def initialize(*responses)
+    @responses = responses
+    @processed_count = 0
+    @cleared = false
+  end
+
+  def read; end
+
+  def process_single_getk_response
+    @processed_count += 1
+    @responses.shift || [nil, nil, nil, nil, nil]
+  end
+
+  def clear
+    @cleared = true
+  end
+
+  def in_progress?
+    !@cleared
+  end
+end
+
+class PipelinedResponseErrorTestConnectionManager
+  attr_reader :finished
+
+  def finish_request!
+    @finished = true
+  end
+
+  def error_on_request!(err)
+    raise err
+  end
+end
+
 describe Dalli::Protocol::Meta::ResponseProcessor do
   it 'includes the full unexpected response line in DalliError messages' do
     # Representative CLIENT_ERROR lines returned by memcached. The response
@@ -73,4 +110,100 @@ def read_line
       assert_equal line.chomp("\r\n"), err.message
     end
   end
+
+  it 'records pipelined error responses instead of treating them as values or terminators' do
+    response_cases = [
+      ["CLIENT_ERROR invalid flag\r\n", Dalli::DalliError, 'Response error: CLIENT_ERROR invalid flag'],
+      [
+        "SERVER_ERROR proxy write to backend failed\r\n",
+        Dalli::ServerError,
+        'SERVER_ERROR proxy write to backend failed'
+      ],
+      ["ERROR\r\n", Dalli::DalliError, 'Response error: ERROR']
+    ]
+
+    processor = Dalli::Protocol::Meta::ResponseProcessor.new(nil, nil)
+
+    response_cases.each do |line, error_class, error_message|
+      bytes, status, cas, key, value, error = processor.getk_response_from_buffer(line)
+
+      assert_equal line.bytesize, bytes
+      refute(status)
+      assert_nil cas
+      assert_nil key
+      assert_nil value
+      assert_instance_of error_class, error
+      assert_equal error_message, error.message
+    end
+  end
+
+  it 'identifies MN as the pipelined noop terminator' do
+    processor = Dalli::Protocol::Meta::ResponseProcessor.new(nil, nil)
+
+    bytes, status, cas, key, value, error = processor.getk_response_from_buffer("MN\r\n")
+
+    assert_equal "MN\r\n".bytesize, bytes
+    assert(status)
+    assert_nil cas
+    assert_nil key
+    assert_nil value
+    assert_nil error
+  end
+
+  it 'raises and drains the pipeline when the first pipelined response is CLIENT_ERROR' do
+    assert_pipelined_error_drained(
+      [false, nil, nil, nil, Dalli::DalliError.new('Response error: CLIENT_ERROR invalid flag')],
+      valid_pipeline_value('a', 'foo'),
+      pipeline_terminator,
+      error_class: Dalli::DalliError,
+      error_message: 'Response error: CLIENT_ERROR invalid flag'
+    )
+  end
+
+  it 'raises and drains the pipeline when a middle pipelined response is SERVER_ERROR' do
+    assert_pipelined_error_drained(
+      valid_pipeline_value('a', 'foo'),
+      [false, nil, nil, nil, Dalli::ServerError.new('SERVER_ERROR proxy write to backend failed')],
+      valid_pipeline_value('b', 'bar'),
+      pipeline_terminator,
+      error_class: Dalli::ServerError,
+      error_message: 'SERVER_ERROR proxy write to backend failed'
+    )
+  end
+
+  it 'raises and drains the pipeline when the last pipelined response is ERROR' do
+    assert_pipelined_error_drained(
+      valid_pipeline_value('a', 'foo'),
+      valid_pipeline_value('b', 'bar'),
+      [false, nil, nil, nil, Dalli::DalliError.new('Response error: ERROR')],
+      pipeline_terminator,
+      error_class: Dalli::DalliError,
+      error_message: 'Response error: ERROR'
+    )
+  end
+
+  def valid_pipeline_value(key, value)
+    [true, nil, key, value, nil]
+  end
+
+  def pipeline_terminator
+    [true, nil, nil, nil, nil]
+  end
+
+  def assert_pipelined_error_drained(*responses, error_class:, error_message:)
+    buffer = PipelinedResponseErrorTestBuffer.new(*responses)
+    connection_manager = PipelinedResponseErrorTestConnectionManager.new
+    server = Dalli::Protocol::Base.allocate
+    server.instance_variable_set(:@response_buffer, buffer)
+    server.instance_variable_set(:@connection_manager, connection_manager)
+
+    err = assert_raises(error_class) do
+      server.pipeline_next_responses
+    end
+
+    assert_equal error_message, err.message
+    assert_equal responses.length, buffer.processed_count
+    assert connection_manager.finished
+    refute_predicate(buffer, :in_progress?)
+  end
 end