diff --git a/staging/src/k8s.io/cli-runtime/pkg/genericclioptions/config_flags.go b/staging/src/k8s.io/cli-runtime/pkg/genericclioptions/config_flags.go
index 754951fd60c..03f4b5332bb 100644
--- a/staging/src/k8s.io/cli-runtime/pkg/genericclioptions/config_flags.go
+++ b/staging/src/k8s.io/cli-runtime/pkg/genericclioptions/config_flags.go
@@ -37,24 +37,25 @@ import (
 )
 
 const (
-	flagClusterName      = "cluster"
-	flagAuthInfoName     = "user"
-	flagContext          = "context"
-	flagNamespace        = "namespace"
-	flagAPIServer        = "server"
-	flagTLSServerName    = "tls-server-name"
-	flagInsecure         = "insecure-skip-tls-verify"
-	flagCertFile         = "client-certificate"
-	flagKeyFile          = "client-key"
-	flagCAFile           = "certificate-authority"
-	flagBearerToken      = "token"
-	flagImpersonate      = "as"
-	flagImpersonateUID   = "as-uid"
-	flagImpersonateGroup = "as-group"
-	flagUsername         = "username"
-	flagPassword         = "password"
-	flagTimeout          = "request-timeout"
-	flagCacheDir         = "cache-dir"
+	flagClusterName        = "cluster"
+	flagAuthInfoName       = "user"
+	flagContext            = "context"
+	flagNamespace          = "namespace"
+	flagAPIServer          = "server"
+	flagTLSServerName      = "tls-server-name"
+	flagInsecure           = "insecure-skip-tls-verify"
+	flagCertFile           = "client-certificate"
+	flagKeyFile            = "client-key"
+	flagCAFile             = "certificate-authority"
+	flagBearerToken        = "token"
+	flagImpersonate        = "as"
+	flagImpersonateUID     = "as-uid"
+	flagImpersonateGroup   = "as-group"
+	flagUsername           = "username"
+	flagPassword           = "password"
+	flagTimeout            = "request-timeout"
+	flagCacheDir           = "cache-dir"
+	flagDisableCompression = "disable-compression"
 )
 
 // RESTClientGetter is an interface that the ConfigFlags describe to provide an easier way to mock for commands
@@ -80,23 +81,24 @@ type ConfigFlags struct {
 	KubeConfig *string
 
 	// config flags
-	ClusterName      *string
-	AuthInfoName     *string
-	Context          *string
-	Namespace        *string
-	APIServer        *string
-	TLSServerName    *string
-	Insecure         *bool
-	CertFile         *string
-	KeyFile          *string
-	CAFile           *string
-	BearerToken      *string
-	Impersonate      *string
-	ImpersonateUID   *string
-	ImpersonateGroup *[]string
-	Username         *string
-	Password         *string
-	Timeout          *string
+	ClusterName        *string
+	AuthInfoName       *string
+	Context            *string
+	Namespace          *string
+	APIServer          *string
+	TLSServerName      *string
+	Insecure           *bool
+	CertFile           *string
+	KeyFile            *string
+	CAFile             *string
+	BearerToken        *string
+	Impersonate        *string
+	ImpersonateUID     *string
+	ImpersonateGroup   *[]string
+	Username           *string
+	Password           *string
+	Timeout            *string
+	DisableCompression *bool
 	// If non-nil, wrap config function can transform the Config
 	// before it is returned in ToRESTConfig function.
 	WrapConfigFn func(*rest.Config) *rest.Config
@@ -199,6 +201,9 @@ func (f *ConfigFlags) toRawKubeConfigLoader() clientcmd.ClientConfig {
 	if f.Insecure != nil {
 		overrides.ClusterInfo.InsecureSkipTLSVerify = *f.Insecure
 	}
+	if f.DisableCompression != nil {
+		overrides.ClusterInfo.DisableCompression = *f.DisableCompression
+	}
 
 	// bind context flags
 	if f.Context != nil {
@@ -393,6 +398,9 @@ func (f *ConfigFlags) AddFlags(flags *pflag.FlagSet) {
 	if f.Timeout != nil {
 		flags.StringVar(f.Timeout, flagTimeout, *f.Timeout, "The length of time to wait before giving up on a single server request. Non-zero values should contain a corresponding time unit (e.g. 1s, 2m, 3h). A value of zero means don't timeout requests.")
 	}
+	if f.DisableCompression != nil {
+		flags.BoolVar(f.DisableCompression, flagDisableCompression, *f.DisableCompression, "If true, opt-out of response compression for all requests to the server")
+	}
 }
 
 // WithDeprecatedPasswordFlag enables the username and password config flags
@@ -424,26 +432,28 @@ func (f *ConfigFlags) WithWrapConfigFn(wrapConfigFn func(*rest.Config) *rest.Con
 func NewConfigFlags(usePersistentConfig bool) *ConfigFlags {
 	impersonateGroup := []string{}
 	insecure := false
+	disableCompression := false
 
 	return &ConfigFlags{
 		Insecure:   &insecure,
 		Timeout:    utilpointer.String("0"),
 		KubeConfig: utilpointer.String(""),
 
-		CacheDir:         utilpointer.String(getDefaultCacheDir()),
-		ClusterName:      utilpointer.String(""),
-		AuthInfoName:     utilpointer.String(""),
-		Context:          utilpointer.String(""),
-		Namespace:        utilpointer.String(""),
-		APIServer:        utilpointer.String(""),
-		TLSServerName:    utilpointer.String(""),
-		CertFile:         utilpointer.String(""),
-		KeyFile:          utilpointer.String(""),
-		CAFile:           utilpointer.String(""),
-		BearerToken:      utilpointer.String(""),
-		Impersonate:      utilpointer.String(""),
-		ImpersonateUID:   utilpointer.String(""),
-		ImpersonateGroup: &impersonateGroup,
+		CacheDir:           utilpointer.String(getDefaultCacheDir()),
+		ClusterName:        utilpointer.String(""),
+		AuthInfoName:       utilpointer.String(""),
+		Context:            utilpointer.String(""),
+		Namespace:          utilpointer.String(""),
+		APIServer:          utilpointer.String(""),
+		TLSServerName:      utilpointer.String(""),
+		CertFile:           utilpointer.String(""),
+		KeyFile:            utilpointer.String(""),
+		CAFile:             utilpointer.String(""),
+		BearerToken:        utilpointer.String(""),
+		Impersonate:        utilpointer.String(""),
+		ImpersonateUID:     utilpointer.String(""),
+		ImpersonateGroup:   &impersonateGroup,
+		DisableCompression: &disableCompression,
 
 		usePersistentConfig: usePersistentConfig,
 		// The more groups you have, the more discovery requests you need to make.
diff --git a/test/cmd/get.sh b/test/cmd/get.sh
index 71d60892a54..644c7f8de0a 100755
--- a/test/cmd/get.sh
+++ b/test/cmd/get.sh
@@ -253,6 +253,48 @@ run_retrieve_multiple_tests() {
   set +o errexit
 }
 
+run_kubectl_response_compression_tests() {
+  set -o nounset
+  set -o errexit
+
+  create_and_use_new_namespace
+  kube::log::status "Testing kubectl get objects with/without response compression"
+
+  ### Test creation of large configmap objects
+  # Pre-condition: no configmaps exists
+  kube::test::get_object_assert configmaps "{{range.items}}{{${id_field:?}}}:{{end}}" ''
+  # Commands to create 3 configmaps each of size 50KB. Sum of their sizes should be >128KB (defaultGzipThresholdBytes)
+  # for apiserver to allow gzip compression of the response. This is required to test the disable-compression option
+  # from client-side
+  some_string=$(dd status=none if=/dev/urandom bs=1K count=50 2>/dev/null | base64)
+  kubectl create configmap "cm-one" --from-literal=somekey="${some_string}"
+  kubectl create configmap "cm-two" --from-literal=somekey="${some_string}"
+  kubectl create configmap "cm-three" --from-literal=somekey="${some_string}"
+  output_message=$(kubectl get configmaps 2>&1 "${kube_flags[@]:?}")
+  # Post-condition: All configmaps should be created
+  kube::test::if_has_string "${output_message}" "cm-one"
+  kube::test::if_has_string "${output_message}" "cm-two"
+  kube::test::if_has_string "${output_message}" "cm-three"
+
+  ### Test list call WITH compression
+  output_message=$(kubectl get configmaps --v=8 2>&1 "${kube_flags[@]:?}")
+  # Post-condition: Response headers should include "accept-encoding" header
+  kube::test::if_has_string "${output_message}" "Vary: Accept-Encoding"
+
+  ### Test list call WITHOUT compression
+  output_message=$(kubectl get configmaps --disable-compression=true --v=8 2>&1 "${kube_flags[@]:?}")
+  # Post-condition: Response headers should NOT include "accept-encoding" header
+  kube::test::if_has_not_string "${output_message}" "Vary: Accept-Encoding"
+
+  # cleanup
+  kubectl delete configmap "cm-one"
+  kubectl delete configmap "cm-two"
+  kubectl delete configmap "cm-three"
+
+  set +o nounset
+  set +o errexit
+}
+
 run_kubectl_sort_by_tests() {
   set -o nounset
   set -o errexit