diff --git a/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource.go b/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource.go
index c978d4bb8d3f..68e755801cbb 100644
--- a/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource.go
+++ b/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource.go
@@ -107,6 +107,12 @@ func resourceHDInsightKafkaCluster() *schema.Resource {
 
 			"storage_account_gen2": SchemaHDInsightsGen2StorageAccounts(),
 
+			"encryption_in_transit_enabled": {
+				Type:     schema.TypeBool,
+				ForceNew: true,
+				Optional: true,
+			},
+
 			"roles": {
 				Type:     schema.TypeList,
 				Required: true,
@@ -251,6 +257,12 @@ func resourceHDInsightKafkaClusterCreate(d *schema.ResourceData, meta interface{
 		Identity: identity,
 	}
 
+	if encryptionInTransit, ok := d.GetOk("encryption_in_transit_enabled"); ok {
+		params.Properties.EncryptionInTransitProperties = &hdinsight.EncryptionInTransitProperties{
+			IsEncryptionInTransitEnabled: utils.Bool(encryptionInTransit.(bool)),
+		}
+	}
+
 	future, err := client.Create(ctx, resourceGroup, name, params)
 	if err != nil {
 		return fmt.Errorf("failure creating HDInsight Kafka Cluster %q (Resource Group %q): %+v", name, resourceGroup, err)
@@ -361,6 +373,10 @@ func resourceHDInsightKafkaClusterRead(d *schema.ResourceData, meta interface{})
 		kafkaRestProxyEndpoint := FindHDInsightConnectivityEndpoint("KafkaRestProxyPublicEndpoint", props.ConnectivityEndpoints)
 		d.Set("kafka_rest_proxy_endpoint", kafkaRestProxyEndpoint)
 
+		if props.EncryptionInTransitProperties != nil {
+			d.Set("encryption_in_transit_enabled", props.EncryptionInTransitProperties.IsEncryptionInTransitEnabled)
+		}
+
 		monitor, err := extensionsClient.GetMonitoringStatus(ctx, resourceGroup, name)
 		if err != nil {
 			return fmt.Errorf("failed reading monitor configuration for HDInsight Hadoop Cluster %q (Resource Group %q): %+v", name, resourceGroup, err)
diff --git a/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource_test.go b/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource_test.go
index 918fe44cb7ac..4c167a3d935a 100644
--- a/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource_test.go
+++ b/azurerm/internal/services/hdinsight/hdinsight_kafka_cluster_resource_test.go
@@ -408,6 +408,28 @@ func TestAccHDInsightKafkaCluster_restProxy(t *testing.T) {
 	})
 }
 
+func TestAccHDInsightKafkaCluster_encryptionInTransitEnabled(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_hdinsight_kafka_cluster", "test")
+	r := HDInsightKafkaClusterResource{}
+	data.ResourceTest(t, r, []resource.TestStep{
+		{
+			Config: r.encryptionInTransitEnabled(data),
+			Check: resource.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep("roles.0.head_node.0.password",
+			"roles.0.head_node.0.vm_size",
+			"roles.0.worker_node.0.password",
+			"roles.0.worker_node.0.vm_size",
+			"roles.0.zookeeper_node.0.password",
+			"roles.0.zookeeper_node.0.vm_size",
+			"roles.0.kafka_management_node.0.password",
+			"roles.0.kafka_management_node.0.vm_size",
+			"storage_account"),
+	})
+}
+
 func (t HDInsightKafkaClusterResource) Exists(ctx context.Context, clients *clients.Client, state *terraform.InstanceState) (*bool, error) {
 	id, err := parse.ClusterID(state.ID)
 	if err != nil {
@@ -1305,3 +1327,57 @@ resource "azurerm_hdinsight_kafka_cluster" "test" {
 }
 `, r.template(data), data.RandomInteger, data.RandomInteger)
 }
+
+func (r HDInsightKafkaClusterResource) encryptionInTransitEnabled(data acceptance.TestData) string {
+	return fmt.Sprintf(`
+%s
+
+resource "azurerm_hdinsight_kafka_cluster" "test" {
+  name                = "acctesthdi-%d"
+  resource_group_name = azurerm_resource_group.test.name
+  location            = azurerm_resource_group.test.location
+  cluster_version     = "4.0"
+  tier                = "Standard"
+
+  encryption_in_transit_enabled = true
+
+  component_version {
+    kafka = "2.1"
+  }
+
+  gateway {
+    enabled  = true
+    username = "acctestusrgw"
+    password = "TerrAform123!"
+  }
+
+  storage_account {
+    storage_container_id = azurerm_storage_container.test.id
+    storage_account_key  = azurerm_storage_account.test.primary_access_key
+    is_default           = true
+  }
+
+  roles {
+    head_node {
+      vm_size  = "Standard_D3_V2"
+      username = "acctestusrvm"
+      password = "AccTestvdSC4daf986!"
+    }
+
+    worker_node {
+      vm_size                  = "Standard_D3_V2"
+      username                 = "acctestusrvm"
+      password                 = "AccTestvdSC4daf986!"
+      target_instance_count    = 3
+      number_of_disks_per_node = 2
+    }
+
+    zookeeper_node {
+      vm_size  = "Standard_D3_V2"
+      username = "acctestusrvm"
+      password = "AccTestvdSC4daf986!"
+    }
+  }
+}
+`, r.template(data), data.RandomInteger)
+}
diff --git a/website/docs/r/hdinsight_kafka_cluster.html.markdown b/website/docs/r/hdinsight_kafka_cluster.html.markdown
index 76218f631cfe..73957d6a8f63 100644
--- a/website/docs/r/hdinsight_kafka_cluster.html.markdown
+++ b/website/docs/r/hdinsight_kafka_cluster.html.markdown
@@ -103,7 +103,9 @@ The following arguments are supported:
 
 * `tier` - (Required) Specifies the Tier which should be used for this HDInsight Kafka Cluster. Possible values are `Standard` or `Premium`. Changing this forces a new resource to be created.
 
-* `min_tls_version` - (Optional) The minimal supported TLS version. Possible values are 1.0, 1.1 or 1.2. Changing this forces a new resource to be created.
+* `min_tls_version` - (Optional) The minimal supported TLS version. Possible values are `1.0`, `1.1` or `1.2`. Changing this forces a new resource to be created.
+
+* `encryption_in_transit_enabled` - (Optional) Whether encryption in transit is enabled for this HDInsight Kafka Cluster. Changing this forces a new resource to be created.
 
 ~> **NOTE:** Starting on June 30, 2020, Azure HDInsight will enforce TLS 1.2 or later versions for all HTTPS connections. For more information, see [Azure HDInsight TLS 1.2 Enforcement](https://azure.microsoft.com/en-us/updates/azure-hdinsight-tls-12-enforcement/).