diff --git a/google/resource_dataproc_cluster.go b/google/resource_dataproc_cluster.go
index 9f4e7f8e..5e780bdf 100644
--- a/google/resource_dataproc_cluster.go
+++ b/google/resource_dataproc_cluster.go
@@ -310,6 +310,19 @@ func resourceDataprocCluster() *schema.Resource {
 								},
 							},
 						},
+						"encryption_config": {
+							Type:     schema.TypeList,
+							Optional: true,
+							MaxItems: 1,
+							Elem: &schema.Resource{
+								Schema: map[string]*schema.Schema{
+									"kms_key_name": {
+										Type:     schema.TypeString,
+										Required: true,
+									},
+								},
+							},
+						},
 					},
 				},
 			},
@@ -502,6 +515,10 @@ func expandClusterConfig(d *schema.ResourceData, config *Config) (*dataproc.Clus
 		conf.InitializationActions = expandInitializationActions(v)
 	}
 
+	if cfg, ok := configOptions(d, "cluster_config.0.encryption_config"); ok {
+		conf.EncryptionConfig = expandEncryptionConfig(cfg)
+	}
+
 	if cfg, ok := configOptions(d, "cluster_config.0.master_config"); ok {
 		log.Println("[INFO] got master_config")
 		conf.MasterConfig = expandInstanceGroupConfig(cfg)
@@ -588,6 +605,14 @@ func expandSoftwareConfig(cfg map[string]interface{}) *dataproc.SoftwareConfig {
 	return conf
 }
 
+func expandEncryptionConfig(cfg map[string]interface{}) *dataproc.EncryptionConfig {
+	conf := &dataproc.EncryptionConfig{}
+	if v, ok := cfg["kms_key_name"]; ok {
+		conf.GcePdKmsKeyName = v.(string)
+	}
+	return conf
+}
+
 func expandInitializationActions(v interface{}) []*dataproc.NodeInitializationAction {
 	actionList := v.([]interface{})
 
@@ -796,6 +821,7 @@ func flattenClusterConfig(d *schema.ResourceData, cfg *dataproc.ClusterConfig) (
 		"master_config":             flattenInstanceGroupConfig(d, cfg.MasterConfig),
 		"worker_config":             flattenInstanceGroupConfig(d, cfg.WorkerConfig),
 		"preemptible_worker_config": flattenPreemptibleInstanceGroupConfig(d, cfg.SecondaryWorkerConfig),
+		"encryption_config":         flattenEncryptionConfig(d, cfg.EncryptionConfig),
 	}
 
 	if len(cfg.InitializationActions) > 0 {
@@ -818,6 +844,14 @@ func flattenSoftwareConfig(d *schema.ResourceData, sc *dataproc.SoftwareConfig)
 	return []map[string]interface{}{data}
 }
 
+func flattenEncryptionConfig(d *schema.ResourceData, ec *dataproc.EncryptionConfig) []map[string]interface{} {
+	data := map[string]interface{}{
+		"kms_key_name": ec.GcePdKmsKeyName,
+	}
+
+	return []map[string]interface{}{data}
+}
+
 func flattenAccelerators(accelerators []*dataproc.AcceleratorConfig) interface{} {
 	acceleratorsTypeSet := schema.NewSet(schema.HashResource(acceleratorsSchema()), []interface{}{})
 	for _, accelerator := range accelerators {
diff --git a/google/resource_dataproc_cluster_test.go b/google/resource_dataproc_cluster_test.go
index 7a57c18b..aa970f19 100644
--- a/google/resource_dataproc_cluster_test.go
+++ b/google/resource_dataproc_cluster_test.go
@@ -481,6 +481,29 @@ func TestAccDataprocCluster_withNetworkRefs(t *testing.T) {
 	})
 }
 
+func TestAccDataprocCluster_KMS(t *testing.T) {
+	t.Parallel()
+
+	rnd := acctest.RandString(10)
+	kms := BootstrapKMSKey(t)
+	pid := getTestProjectFromEnv()
+
+	var cluster dataproc.Cluster
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckDataprocClusterDestroy(),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataprocCluster_KMS(pid, rnd, kms.CryptoKey.Name),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDataprocClusterExists("google_dataproc_cluster.kms", &cluster),
+				),
+			},
+		},
+	})
+}
+
 func testAccCheckDataprocClusterDestroy() resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		config := testAccProvider.Meta().(*Config)
@@ -777,7 +800,7 @@ resource "google_dataproc_cluster" "basic" {
 	name                  = "dproc-cluster-test-%s"
 	region                = "us-central1"
 	depends_on            = ["google_compute_firewall.dataproc_network_firewall"]
-	
+
 	cluster_config {
 		gce_cluster_config {
 			subnetwork       = "${google_compute_subnetwork.dataproc_subnetwork.name}"
@@ -1155,3 +1178,27 @@ resource "google_dataproc_cluster" "with_net_ref_by_url" {
 
 `, netName, rnd, rnd, rnd)
 }
+
+func testAccDataprocCluster_KMS(pid, rnd, kmsKey string) string {
+	return fmt.Sprintf(`
+data "google_project" "project" {
+	project_id = "%s"
+}
+
+resource "google_project_iam_member" "kms-project-binding" {
+  project = "${data.google_project.project.project_id}"
+	role    = "roles/cloudkms.cryptoKeyEncrypterDecrypter"
+	member  = "serviceAccount:service-${data.google_project.project.number}@compute-system.iam.gserviceaccount.com"
+}
+
+resource "google_dataproc_cluster" "kms" {
+	name   = "dproc-cluster-test-%s"
+	region = "us-central1"
+
+	cluster_config {
+		encryption_config {
+			kms_key_name = "%s"
+		}
+	}
+}`, pid, rnd, kmsKey)
+}
diff --git a/website/docs/r/dataproc_cluster.html.markdown b/website/docs/r/dataproc_cluster.html.markdown
index e15e1538..1a1f9fb3 100644
--- a/website/docs/r/dataproc_cluster.html.markdown
+++ b/website/docs/r/dataproc_cluster.html.markdown
@@ -139,6 +139,7 @@ The `cluster_config` block supports:
 
         # You can define multiple initialization_action blocks
         initialization_action     { ... }
+        encryption_config         { ... }
     }
 ```
 
@@ -168,6 +169,8 @@ The `cluster_config` block supports:
 * `initialization_action` (Optional) Commands to execute on each node after config is completed.
    You can specify multiple versions of these. Structure defined below.
 
+* `encryption_config` (Optional) The Customer managed encryption keys settings for the cluster.
+   Structure defined below.
 - - -
 
 The `cluster_config.gce_cluster_config` block supports:
@@ -418,6 +421,22 @@ The `initialization_action` block (Optional) can be specified multiple times and
    allowed to take to execute its action. GCP will default to a predetermined
    computed value if not set (currently 300).
 
+- - -
+
+The `encryption_config` block supports:
+
+```hcl
+    cluster_config {
+        encryption_config {
+            kms_key_name = "projects/projectId/locations/region/keyRings/keyRingName/cryptoKeys/keyName"
+        }
+    }
+}
+```
+
+* `kms_key_name` - (Required) The Cloud KMS key name to use for PD disk encryption for
+   all instances in the cluster.
+
 ## Attributes Reference
 
 In addition to the arguments listed above, the following computed attributes are