diff --git a/pkg/cassandra/config/config.go b/pkg/cassandra/config/config.go
index 3f5d1c2eff9..0ce19401a27 100644
--- a/pkg/cassandra/config/config.go
+++ b/pkg/cassandra/config/config.go
@@ -38,6 +38,7 @@ type Configuration struct {
 	MaxRetryAttempts     int           `validate:"min=0" yaml:"max_retry_attempt"`
 	ProtoVersion         int           `yaml:"proto_version"`
 	Consistency          string        `yaml:"consistency"`
+	DisableCompression   bool          `yaml:"disable-compression"`
 	Port                 int           `yaml:"port"`
 	Authenticator        Authenticator `yaml:"authenticator"`
 	DisableAutoDiscovery bool          `yaml:"disable_auto_discovery"`
@@ -128,7 +129,11 @@ func (c *Configuration) NewCluster() *gocql.ClusterConfig {
 	if c.Port != 0 {
 		cluster.Port = c.Port
 	}
-	cluster.Compressor = gocql.SnappyCompressor{}
+
+	if !c.DisableCompression {
+		cluster.Compressor = gocql.SnappyCompressor{}
+	}
+
 	if c.Consistency == "" {
 		cluster.Consistency = gocql.LocalOne
 	} else {
diff --git a/plugin/storage/cassandra/options.go b/plugin/storage/cassandra/options.go
index fa61def2081..44f3ca66691 100644
--- a/plugin/storage/cassandra/options.go
+++ b/plugin/storage/cassandra/options.go
@@ -37,6 +37,7 @@ const (
 	suffixKeyspace             = ".keyspace"
 	suffixDC                   = ".local-dc"
 	suffixConsistency          = ".consistency"
+	suffixDisableCompression   = ".disable-compression"
 	suffixProtoVer             = ".proto-version"
 	suffixSocketKeepAlive      = ".socket-keep-alive"
 	suffixUsername             = ".username"
@@ -163,6 +164,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixConsistency,
 		nsConfig.Consistency,
 		"The Cassandra consistency level, e.g. ANY, ONE, TWO, THREE, QUORUM, ALL, LOCAL_QUORUM, EACH_QUORUM, LOCAL_ONE (default LOCAL_ONE)")
+	flagSet.Bool(
+		nsConfig.namespace+suffixDisableCompression,
+		false,
+		"Disables the use of the default Snappy Compression while connecting to the Cassandra Cluster if set to true. This is useful for connecting to Cassandra Clusters(like Azure Cosmos Db with Cassandra API) that do not support SnappyCompression")
 	flagSet.Int(
 		nsConfig.namespace+suffixProtoVer,
 		nsConfig.ProtoVersion,
@@ -243,6 +248,7 @@ func (cfg *namespaceConfig) initFromViper(v *viper.Viper) {
 	cfg.TLS.ServerName = v.GetString(cfg.namespace + suffixServerName)
 	cfg.TLS.EnableHostVerification = v.GetBool(cfg.namespace + suffixVerifyHost)
 	cfg.EnableDependenciesV2 = v.GetBool(cfg.namespace + suffixEnableDependenciesV2)
+	cfg.DisableCompression = v.GetBool(cfg.namespace + suffixDisableCompression)
 }
 
 // GetPrimary returns primary configuration.