diff --git a/contrib/Shopify/sarama/option.go b/contrib/Shopify/sarama/option.go
index f2443b6a82..1c574d9680 100644
--- a/contrib/Shopify/sarama/option.go
+++ b/contrib/Shopify/sarama/option.go
@@ -34,6 +34,8 @@ func defaults(cfg *config) {
 	cfg.consumerSpanName = namingschema.NewKafkaInboundOp().GetName()
 	cfg.producerSpanName = namingschema.NewKafkaOutboundOp().GetName()
 
+	cfg.dataStreamsEnabled = internal.BoolEnv("DD_DATA_STREAMS_ENABLED", false)
+
 	// cfg.analyticsRate = globalconfig.AnalyticsRate()
 	if internal.BoolEnv("DD_TRACE_SARAMA_ANALYTICS_ENABLED", false) {
 		cfg.analyticsRate = 1.0
diff --git a/contrib/Shopify/sarama/option_test.go b/contrib/Shopify/sarama/option_test.go
new file mode 100644
index 0000000000..d2bba4fed8
--- /dev/null
+++ b/contrib/Shopify/sarama/option_test.go
@@ -0,0 +1,38 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package sarama
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestDataStreamsActivation(t *testing.T) {
+	t.Run("default", func(t *testing.T) {
+		cfg := new(config)
+		defaults(cfg)
+		assert.False(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withOption", func(t *testing.T) {
+		cfg := new(config)
+		defaults(cfg)
+		WithDataStreams()(cfg)
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
+		cfg := new(config)
+		defaults(cfg)
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("optionOverridesEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "false")
+		cfg := new(config)
+		defaults(cfg)
+		WithDataStreams()(cfg)
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+}
diff --git a/contrib/confluentinc/confluent-kafka-go/kafka.v2/option.go b/contrib/confluentinc/confluent-kafka-go/kafka.v2/option.go
index d946a426ed..767686957a 100644
--- a/contrib/confluentinc/confluent-kafka-go/kafka.v2/option.go
+++ b/contrib/confluentinc/confluent-kafka-go/kafka.v2/option.go
@@ -41,6 +41,7 @@ func newConfig(opts ...Option) *config {
 		// analyticsRate: globalconfig.AnalyticsRate(),
 		analyticsRate: math.NaN(),
 	}
+	cfg.dataStreamsEnabled = internal.BoolEnv("DD_DATA_STREAMS_ENABLED", false)
 	if internal.BoolEnv("DD_TRACE_KAFKA_ANALYTICS_ENABLED", false) {
 		cfg.analyticsRate = 1.0
 	}
diff --git a/contrib/confluentinc/confluent-kafka-go/kafka.v2/option_test.go b/contrib/confluentinc/confluent-kafka-go/kafka.v2/option_test.go
index c97e149801..d990870fc5 100644
--- a/contrib/confluentinc/confluent-kafka-go/kafka.v2/option_test.go
+++ b/contrib/confluentinc/confluent-kafka-go/kafka.v2/option_test.go
@@ -14,6 +14,27 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestDataStreamsActivation(t *testing.T) {
+	t.Run("default", func(t *testing.T) {
+		cfg := newConfig()
+		assert.False(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withOption", func(t *testing.T) {
+		cfg := newConfig(WithDataStreams())
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
+		cfg := newConfig()
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("optionOverridesEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "false")
+		cfg := newConfig(WithDataStreams())
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+}
+
 func TestAnalyticsSettings(t *testing.T) {
 	t.Run("defaults", func(t *testing.T) {
 		cfg := newConfig()
diff --git a/contrib/confluentinc/confluent-kafka-go/kafka/option.go b/contrib/confluentinc/confluent-kafka-go/kafka/option.go
index 514f54fff0..aeb1526ed7 100644
--- a/contrib/confluentinc/confluent-kafka-go/kafka/option.go
+++ b/contrib/confluentinc/confluent-kafka-go/kafka/option.go
@@ -41,6 +41,7 @@ func newConfig(opts ...Option) *config {
 		// analyticsRate: globalconfig.AnalyticsRate(),
 		analyticsRate: math.NaN(),
 	}
+	cfg.dataStreamsEnabled = internal.BoolEnv("DD_DATA_STREAMS_ENABLED", false)
 	if internal.BoolEnv("DD_TRACE_KAFKA_ANALYTICS_ENABLED", false) {
 		cfg.analyticsRate = 1.0
 	}
diff --git a/contrib/confluentinc/confluent-kafka-go/kafka/option_test.go b/contrib/confluentinc/confluent-kafka-go/kafka/option_test.go
index c97e149801..d990870fc5 100644
--- a/contrib/confluentinc/confluent-kafka-go/kafka/option_test.go
+++ b/contrib/confluentinc/confluent-kafka-go/kafka/option_test.go
@@ -14,6 +14,27 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestDataStreamsActivation(t *testing.T) {
+	t.Run("default", func(t *testing.T) {
+		cfg := newConfig()
+		assert.False(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withOption", func(t *testing.T) {
+		cfg := newConfig(WithDataStreams())
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("withEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "true")
+		cfg := newConfig()
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+	t.Run("optionOverridesEnv", func(t *testing.T) {
+		t.Setenv("DD_DATA_STREAMS_ENABLED", "false")
+		cfg := newConfig(WithDataStreams())
+		assert.True(t, cfg.dataStreamsEnabled)
+	})
+}
+
 func TestAnalyticsSettings(t *testing.T) {
 	t.Run("defaults", func(t *testing.T) {
 		cfg := newConfig()