diff --git a/azurerm/internal/services/kusto/kusto_cluster_resource.go b/azurerm/internal/services/kusto/kusto_cluster_resource.go
index 4fe236344b78..fe194c0b0029 100644
--- a/azurerm/internal/services/kusto/kusto_cluster_resource.go
+++ b/azurerm/internal/services/kusto/kusto_cluster_resource.go
@@ -94,6 +94,11 @@ func resourceArmKustoCluster() *schema.Resource {
 				Optional: true,
 			},
 
+			"enable_purge": {
+				Type:     schema.TypeBool,
+				Optional: true,
+			},
+
 			"uri": {
 				Type:     schema.TypeString,
 				Computed: true,
@@ -142,6 +147,7 @@ func resourceArmKustoClusterCreateUpdate(d *schema.ResourceData, meta interface{
 	clusterProperties := kusto.ClusterProperties{
 		EnableDiskEncryption:  utils.Bool(d.Get("enable_disk_encryption").(bool)),
 		EnableStreamingIngest: utils.Bool(d.Get("enable_streaming_ingest").(bool)),
+		EnablePurge:           utils.Bool(d.Get("enable_purge").(bool)),
 	}
 
 	t := d.Get("tags").(map[string]interface{})
@@ -211,6 +217,7 @@ func resourceArmKustoClusterRead(d *schema.ResourceData, meta interface{}) error
 	if clusterProperties := clusterResponse.ClusterProperties; clusterProperties != nil {
 		d.Set("enable_disk_encryption", clusterProperties.EnableDiskEncryption)
 		d.Set("enable_streaming_ingest", clusterProperties.EnableStreamingIngest)
+		d.Set("enable_purge", clusterProperties.EnablePurge)
 		d.Set("uri", clusterProperties.URI)
 		d.Set("data_ingestion_uri", clusterProperties.DataIngestionURI)
 	}
diff --git a/azurerm/internal/services/kusto/tests/kusto_cluster_resource_test.go b/azurerm/internal/services/kusto/tests/kusto_cluster_resource_test.go
index 0de371fb4793..ff6009d0ece4 100644
--- a/azurerm/internal/services/kusto/tests/kusto_cluster_resource_test.go
+++ b/azurerm/internal/services/kusto/tests/kusto_cluster_resource_test.go
@@ -44,6 +44,7 @@ func TestAccAzureRMKustoCluster_update(t *testing.T) {
 					testCheckAzureRMKustoClusterExists(data.ResourceName),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_disk_encryption", "false"),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_streaming_ingest", "false"),
+					resource.TestCheckResourceAttr(data.ResourceName, "enable_purge", "false"),
 				),
 			},
 			data.ImportStep(),
@@ -53,6 +54,7 @@ func TestAccAzureRMKustoCluster_update(t *testing.T) {
 					testCheckAzureRMKustoClusterExists(data.ResourceName),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_disk_encryption", "true"),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_streaming_ingest", "true"),
+					resource.TestCheckResourceAttr(data.ResourceName, "enable_purge", "true"),
 				),
 			},
 			data.ImportStep(),
@@ -62,6 +64,7 @@ func TestAccAzureRMKustoCluster_update(t *testing.T) {
 					testCheckAzureRMKustoClusterExists(data.ResourceName),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_disk_encryption", "false"),
 					resource.TestCheckResourceAttr(data.ResourceName, "enable_streaming_ingest", "false"),
+					resource.TestCheckResourceAttr(data.ResourceName, "enable_purge", "false"),
 				),
 			},
 			data.ImportStep(),
@@ -248,6 +251,7 @@ resource "azurerm_kusto_cluster" "test" {
   resource_group_name     = azurerm_resource_group.test.name
   enable_disk_encryption  = true
   enable_streaming_ingest = true
+  enable_purge            = true
 
   sku {
     name     = "Dev(No SLA)_Standard_D11_v2"
diff --git a/website/docs/r/kusto_cluster.html.markdown b/website/docs/r/kusto_cluster.html.markdown
index 330a8d0febc5..234719b4dbfc 100644
--- a/website/docs/r/kusto_cluster.html.markdown
+++ b/website/docs/r/kusto_cluster.html.markdown
@@ -50,6 +50,8 @@ The following arguments are supported:
 
 * `enable_streaming_ingest` - (Optional) Specifies if the streaming ingest is enabled.
 
+* `enable_purge` - (Optional) Specifies if the purge operations are enabled.
+
 * `tags` - (Optional) A mapping of tags to assign to the resource.
 
 ---