diff --git a/kong/pdk/service/response.lua b/kong/pdk/service/response.lua
index 5a6621abf543..3fe0407a1ab1 100644
--- a/kong/pdk/service/response.lua
+++ b/kong/pdk/service/response.lua
@@ -36,90 +36,25 @@ local header_body_log = phase_checker.new(PHASES.response,
                                           PHASES.body_filter,
                                           PHASES.log)
 
-
-local attach_resp_headers_mt
-
+local clear_non_upstream_headers
 
 do
-  local resp_headers_orig_mt_index
-
-
-  local resp_headers_mt = {
-    __index = function(t, name)
-      if type(name) == "string" then
-        local var = fmt("upstream_http_%s", replace_dashes_lower(name))
-        if not ngx.var[var] then
-          return nil
-        end
+  -- pairs in a hot code path - although this hits an NYI and falls back to the
+  -- interpreter, it uses an optimized hand-written assembly form of loop on table
+  -- that does not use `next`
+  clear_non_upstream_headers = function (response_headers, err)
+    for header_name, _ in pairs(response_headers) do
+      local var = fmt("upstream_http_%s", replace_dashes_lower(header_name))
+      if not ngx.var[var] then
+        response_headers[header_name] = nil
       end
-
-      return resp_headers_orig_mt_index(t, name)
-    end,
-  }
-
-
-  attach_resp_headers_mt = function(response_headers, err)
-    if not resp_headers_orig_mt_index then
-      local mt = getmetatable(response_headers)
-      resp_headers_orig_mt_index = mt.__index
     end
 
-    setmetatable(response_headers, resp_headers_mt)
-
     return response_headers, err
   end
 end
 
 
-local attach_buffered_headers_mt
-
-do
-  local EMPTY = {}
-
-  attach_buffered_headers_mt = function(response_headers, max_headers)
-    if not response_headers then
-      return EMPTY
-    end
-
-    return setmetatable({}, { __index = function(_, name)
-      if type(name) ~= "string" then
-        return nil
-      end
-
-      if response_headers[name] then
-        return response_headers[name]
-      end
-
-      name = lower(name)
-
-      if response_headers[name] then
-        return response_headers[name]
-      end
-
-      name = replace_dashes(name)
-
-      if response_headers[name] then
-        return response_headers[name]
-      end
-
-      local i = 1
-      for n, v in pairs(response_headers) do
-        if i > max_headers then
-          return nil
-        end
-
-        n = replace_dashes_lower(n)
-        if n == name then
-          return v
-        end
-
-        i = i + 1
-      end
-    end })
-  end
-end
-
-
 local function new(pdk, major_version)
   local response = {}
 
@@ -137,8 +72,8 @@ local function new(pdk, major_version)
   local MIN_HEADERS            = 1
   local MAX_HEADERS_DEFAULT    = 100
   local MAX_HEADERS            = 1000
-  local MAX_HEADERS_CONFIGURED
-
+  local MAX_HEADERS_CONFIGURED = pdk.configuration and pdk.configuration.lua_max_resp_headers
+                                  or MAX_HEADERS_DEFAULT
 
   ---
   -- Returns the HTTP status code of the response from the Service as a Lua number.
@@ -196,39 +131,27 @@ local function new(pdk, major_version)
   --   kong.log.inspect(headers.x_another[1])    -- "foo bar"
   --   kong.log.inspect(headers["X-Another"][2]) -- "baz"
   -- end
-  -- Note that this function returns a proxy table
-  -- which cannot be iterated with `pairs` or used as operand of `#`.
   function response.get_headers(max_headers)
     check_phase(header_body_log)
 
-    local buffered_headers = ngx.ctx.buffered_headers
-
-    if max_headers == nil then
-      if buffered_headers then
-        if not MAX_HEADERS_CONFIGURED then
-          MAX_HEADERS_CONFIGURED = pdk and pdk.configuration and pdk.configuration.lua_max_resp_headers
-        end
-        return attach_buffered_headers_mt(buffered_headers, MAX_HEADERS_CONFIGURED or MAX_HEADERS_DEFAULT)
-      end
-
-      return attach_resp_headers_mt(ngx.resp.get_headers())
-    end
+    if max_headers ~= nil and type(max_headers) ~= "number" then
+      error("max_headers must be a number or nil", 2)
 
-    if type(max_headers) ~= "number" then
-      error("max_headers must be a number", 2)
-
-    elseif max_headers < MIN_HEADERS then
+    elseif max_headers and max_headers < MIN_HEADERS then
       error("max_headers must be >= " .. MIN_HEADERS, 2)
 
-    elseif max_headers > MAX_HEADERS then
+    elseif max_headers and max_headers > MAX_HEADERS then
       error("max_headers must be <= " .. MAX_HEADERS, 2)
     end
 
-    if buffered_headers then
-      return attach_buffered_headers_mt(buffered_headers, max_headers)
+    local ctx = ngx.ctx
+    if ctx.buffered_proxying and ctx.buffered_headers then
+      -- XXX ensure metatable is the same as for non-buffered
+      return ctx.buffered_headers
     end
 
-    return attach_resp_headers_mt(ngx.resp.get_headers(max_headers))
+    -- we patch ngx.resp.get_headers to honor the max headers kong config
+    return clear_non_upstream_headers(ngx.resp.get_headers(max_headers))
   end
 
   ---
diff --git a/t/01-pdk/07-service-response/02-get_headers.t b/t/01-pdk/07-service-response/02-get_headers.t
index 2a5042f46474..f3f88fc4101e 100644
--- a/t/01-pdk/07-service-response/02-get_headers.t
+++ b/t/01-pdk/07-service-response/02-get_headers.t
@@ -11,7 +11,359 @@ run_tests();
 
 __DATA__
 
-=== TEST 1: service.response.get_headers() returns a table
+=== TEST 1: service.response.get_headers() returns an iterable table (non-buffered)
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+                ngx.header.Foo = "Hello"
+            }
+        }
+    }
+}
+--- config
+    location /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers()
+            ngx.arg[1] = "type: " .. type(headers) .. "\nfirst: " .. (next(headers) and "exists" or "nil")
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+type: table
+first: exists
+--- no_error_log
+[error]
+
+
+
+=== TEST 2: service.response.get_headers() returns an iterable table (buffered)
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+                ngx.header.Foo = "Hello"
+            }
+        }
+    }
+}
+--- config
+    location /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        access_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+            pdk.service.request.enable_buffering()
+            ngx.ctx.buffered_headers = {
+                ["Foo"] = "Hello",
+                ["Bar"] = "World",
+                ["Accept"] = {
+                    "application/json",
+                    "text/html",
+                }
+            }
+
+            ngx.header.h = "oops, i am here"
+        }
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers()
+            ngx.arg[1] = "type: " .. type(headers) .. "\nfirst: " .. (next(headers) and "exists" or "nil")
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+type: table
+first: exists
+--- no_error_log
+[error]
+
+
+
+=== TEST 3: service.response.get_headers() returns service response headers only (unbuffered mode) BROKEN
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+                ngx.header.Foo = "Hello"
+                ngx.header.Bar = "World"
+                ngx.header.Accept = {
+                    "application/json",
+                    "text/html",
+                }
+            }
+        }
+    }
+}
+--- config
+    location = /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        access_by_lua_block {
+            ngx.header.h = "oops, i am here"
+        }
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers()
+
+            ngx.arg[1] = "Foo: " .. headers.Foo .. "\n" ..
+                         "Bar: " .. headers.Bar .. "\n" ..
+                         "Accept: " .. table.concat(headers.Accept, ", ") .. "\n" ..
+                         "NonUpstreamHeader: " .. (headers.h or "i must not be here") .. "\n" ..
+                         "Buffered: " .. (ngx.ctx.buffered_proxying and "true" or "false")
+
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+Foo: Hello
+Bar: World
+Accept: application/json, text/html
+NonUpstreamHeader: i must not be here
+Buffered: false
+--- no_error_log
+[error]
+
+
+
+=== TEST 4: service.response.get_headers() returns service response headers only (unbuffered mode, max_headers) BROKEN
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+                ngx.header.Foo = "Hello"
+                ngx.header.Bar = "World"
+                ngx.header.Accept = {
+                    "application/json",
+                    "text/html",
+                }
+            }
+        }
+    }
+}
+--- config
+    location = /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        access_by_lua_block {
+            ngx.header.h = "oops, i am here"
+        }
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers(10)
+
+            ngx.arg[1] = "Foo: " .. headers.Foo .. "\n" ..
+                         "Bar: " .. headers.Bar .. "\n" ..
+                         "Accept: " .. table.concat(headers.Accept, ", ") .. "\n" ..
+                         "NonUpstreamHeader: " .. (headers.h or "i must not be here") .. "\n" ..
+                         "Buffered: " .. (ngx.ctx.buffered_proxying and "true" or "false")
+
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+Foo: Hello
+Bar: World
+Accept: application/json, text/html
+NonUpstreamHeader: i must not be here
+Buffered: false
+--- no_error_log
+[error]
+
+
+
+=== TEST 5: service.response.get_headers() returns service response headers (buffered mode) CORRECT
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+            }
+        }
+    }
+}
+--- config
+    location = /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        access_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+            pdk.service.request.enable_buffering()
+            ngx.ctx.buffered_headers = {
+                ["Foo"] = "Hello",
+                ["Bar"] = "World",
+                ["Accept"] = {
+                    "application/json",
+                    "text/html",
+                }
+            }
+
+            ngx.header.h = "oops, i am here"
+        }
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers()
+
+            ngx.arg[1] = "Foo: " .. headers.Foo .. "\n" ..
+                         "Bar: " .. headers.Bar .. "\n" ..
+                         "Accept: " .. table.concat(headers.Accept, ", ") .. "\n" ..
+                         "NonUpstreamHeader: " .. (headers.h or "i must not be here") .. "\n" ..
+                         "Buffered: " .. (ngx.ctx.buffered_proxying and "true" or "false")
+
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+Foo: Hello
+Bar: World
+Accept: application/json, text/html
+NonUpstreamHeader: i must not be here
+Buffered: true
+--- no_error_log
+[error]
+
+
+
+=== TEST 6: service.response.get_headers() returns service response headers (buffered mode, max_headers) CORRECT
+--- http_config eval
+qq{
+    $t::Util::HttpConfig
+
+    server {
+        listen unix:$ENV{TEST_NGINX_NXSOCK}/nginx.sock;
+
+        location / {
+            content_by_lua_block {
+            }
+        }
+    }
+}
+--- config
+    location = /t {
+        proxy_pass http://unix:$TEST_NGINX_NXSOCK/nginx.sock;
+
+        access_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+            pdk.service.request.enable_buffering()
+            ngx.ctx.buffered_headers = {
+                ["Foo"] = "Hello",
+                ["Bar"] = "World",
+                ["Accept"] = {
+                    "application/json",
+                    "text/html",
+                }
+            }
+
+            ngx.header.h = "oops, i am here"
+        }
+
+        header_filter_by_lua_block {
+            ngx.header.content_length = nil
+        }
+
+        body_filter_by_lua_block {
+            local PDK = require "kong.pdk"
+            local pdk = PDK.new()
+
+            local headers = pdk.service.response.get_headers(10)
+
+            ngx.arg[1] = "Foo: " .. headers.Foo .. "\n" ..
+                         "Bar: " .. headers.Bar .. "\n" ..
+                         "Accept: " .. table.concat(headers.Accept, ", ") .. "\n" ..
+                         "NonUpstreamHeader: " .. (headers.h or "i must not be here") .. "\n" ..
+                         "Buffered: " .. (ngx.ctx.buffered_proxying and "true" or "false")
+
+            ngx.arg[2] = true
+        }
+    }
+--- request
+GET /t
+--- response_body chop
+Foo: Hello
+Bar: World
+Accept: application/json, text/html
+NonUpstreamHeader: i must not be here
+Buffered: true
+--- no_error_log
+[error]
+
+
+
+=== TEST 7: service.response.get_headers() returns a table
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -49,7 +401,7 @@ type: table
 
 
 
-=== TEST 2: service.response.get_headers() returns service response headers
+=== TEST 8: service.response.get_headers() returns service response headers
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -102,7 +454,7 @@ Accept: application/json, text/html
 
 
 
-=== TEST 3: service.response.get_headers() returns service response headers with case-insensitive metatable
+=== TEST 9: service.response.get_headers() returns service response headers with case-insensitive metatable
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -151,7 +503,7 @@ x_Foo_header: Hello
 
 
 
-=== TEST 4: service.response.get_headers() fetches 100 headers max by default
+=== TEST 10: service.response.get_headers() fetches 100 headers max by default
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -190,20 +542,20 @@ qq{
                 n = n + 1
             end
 
-            ngx.arg[1] = ngx.arg[1] .. "number of headers fetched: " .. n
+            ngx.arg[1] = ngx.arg[1] .. "number of headers fetched <= 100? " .. (n <= 100 and "true" or "false")
             ngx.arg[2] = true
         }
     }
 --- request
 GET /t
 --- response_body chop
-number of headers fetched: 100
+number of headers fetched <= 100? true
 --- no_error_log
 [error]
 
 
 
-=== TEST 5: service.response.get_headers() returns error when truncating
+=== TEST 11: service.response.get_headers() returns error when truncating
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -250,7 +602,7 @@ truncated
 
 
 
-=== TEST 6: service.response.get_headers() fetches max_headers argument
+=== TEST 12: service.response.get_headers() fetches max_headers argument
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -281,7 +633,8 @@ qq{
             local PDK = require "kong.pdk"
             local pdk = PDK.new()
 
-            local headers = pdk.service.response.get_headers(60)
+            local max_headers = 60
+            local headers = pdk.service.response.get_headers(max_headers)
 
             local n = 0
 
@@ -289,20 +642,20 @@ qq{
                 n = n + 1
             end
 
-            ngx.arg[1] = ngx.arg[1] .. "number of headers fetched: " .. n
+            ngx.arg[1] = ngx.arg[1] .. "number of headers fetched <= max_headers? " .. (n <= max_headers and "true" or "false")
             ngx.arg[2] = true
         }
     }
 --- request
 GET /t
 --- response_body chop
-number of headers fetched: 60
+number of headers fetched <= max_headers? true
 --- no_error_log
 [error]
 
 
 
-=== TEST 7: service.response.get_headers() raises error when trying to fetch with max_headers invalid value
+=== TEST 13: service.response.get_headers() raises error when trying to fetch with max_headers invalid value
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -336,13 +689,13 @@ qq{
 --- request
 GET /t
 --- response_body chop
-error: max_headers must be a number
+error: max_headers must be a number or nil
 --- no_error_log
 [error]
 
 
 
-=== TEST 8: service.response.get_headers() raises error when trying to fetch with max_headers < 1
+=== TEST 14: service.response.get_headers() raises error when trying to fetch with max_headers < 1
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -382,7 +735,7 @@ error: max_headers must be >= 1
 
 
 
-=== TEST 9: service.response.get_headers() raises error when trying to fetch with max_headers > 1000
+=== TEST 15: service.response.get_headers() raises error when trying to fetch with max_headers > 1000
 --- http_config eval
 qq{
     $t::Util::HttpConfig
@@ -422,7 +775,7 @@ error: max_headers must be <= 1000
 
 
 
-=== TEST 10: service.response.get_headers() returns only service headers
+=== TEST 16: service.response.get_headers() returns only service headers
 --- http_config eval
 qq{
     $t::Util::HttpConfig