diff --git a/.changelog/4359.txt b/.changelog/4359.txt
new file mode 100644
index 0000000000..a513d01258
--- /dev/null
+++ b/.changelog/4359.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+dataflow: Added optional `kms_key_name` field for `google_dataflow_job`
+```
diff --git a/google-beta/resource_dataflow_job.go b/google-beta/resource_dataflow_job.go
index 928042a4a4..79e7331e9f 100644
--- a/google-beta/resource_dataflow_job.go
+++ b/google-beta/resource_dataflow_job.go
@@ -171,6 +171,12 @@ func resourceDataflowJob() *schema.Resource {
 				Description: `The machine type to use for the job.`,
 			},
 
+			"kms_key_name": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Description: `The name for the Cloud KMS key for the job. Key format is: projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY`,
+			},
+
 			"ip_configuration": {
 				Type:         schema.TypeString,
 				Optional:     true,
@@ -303,6 +309,9 @@ func resourceDataflowJobRead(d *schema.ResourceData, meta interface{}) error {
 	if err := d.Set("labels", job.Labels); err != nil {
 		return fmt.Errorf("Error setting labels: %s", err)
 	}
+	if err := d.Set("kms_key_name", job.Environment.ServiceKmsKeyName); err != nil {
+		return fmt.Errorf("Error setting kms_key_name: %s", err)
+	}
 
 	sdkPipelineOptions, err := ConvertToMap(job.Environment.SdkPipelineOptions)
 	if err != nil {
@@ -529,6 +538,7 @@ func resourceDataflowJobSetupEnv(d *schema.ResourceData, config *Config) (datafl
 		Subnetwork:            d.Get("subnetwork").(string),
 		TempLocation:          d.Get("temp_gcs_location").(string),
 		MachineType:           d.Get("machine_type").(string),
+		KmsKeyName:            d.Get("kms_key_name").(string),
 		IpConfiguration:       d.Get("ip_configuration").(string),
 		AdditionalUserLabels:  labels,
 		Zone:                  zone,
diff --git a/google-beta/resource_dataflow_job_test.go b/google-beta/resource_dataflow_job_test.go
index 80e4b39bf4..677fb85295 100644
--- a/google-beta/resource_dataflow_job_test.go
+++ b/google-beta/resource_dataflow_job_test.go
@@ -204,6 +204,33 @@ func TestAccDataflowJob_withIpConfig(t *testing.T) {
 	})
 }
 
+func TestAccDataflowJob_withKmsKey(t *testing.T) {
+	// Dataflow responses include serialized java classes and bash commands
+	// This makes body comparison infeasible
+	skipIfVcr(t)
+	t.Parallel()
+
+	randStr := randString(t, 10)
+	key_ring := "tf-test-dataflow-kms-ring-" + randStr
+	crypto_key := "tf-test-dataflow-kms-key-" + randStr
+	bucket := "tf-test-dataflow-gcs-" + randStr
+	job := "tf-test-dataflow-job-" + randStr
+	zone := "us-central1-f"
+
+	vcrTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckDataflowJobDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDataflowJob_kms(key_ring, crypto_key, bucket, job, zone),
+				Check: resource.ComposeTestCheckFunc(
+					testAccDataflowJobExists(t, "google_dataflow_job.big_data"),
+				),
+			},
+		},
+	})
+}
 func TestAccDataflowJobWithAdditionalExperiments(t *testing.T) {
 	// Dataflow responses include serialized java classes and bash commands
 	// This makes body comparison infeasible
@@ -783,6 +810,57 @@ resource "google_dataflow_job" "with_labels" {
 
 }
 
+func testAccDataflowJob_kms(key_ring, crypto_key, bucket, job, zone string) string {
+	return fmt.Sprintf(`
+data "google_project" "project" {
+}
+
+resource "google_project_iam_member" "kms-project-dataflow-binding" {
+  project = data.google_project.project.project_id
+  role    = "roles/cloudkms.cryptoKeyEncrypterDecrypter"
+  member  = "serviceAccount:service-${data.google_project.project.number}@dataflow-service-producer-prod.iam.gserviceaccount.com"
+}
+
+resource "google_project_iam_member" "kms-project-compute-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_kms_key_ring" "keyring" {
+  name     = "%s"
+  location = "global"
+}
+
+resource "google_kms_crypto_key" "crypto_key" {
+  name            = "%s"
+  key_ring        = google_kms_key_ring.keyring.id
+  rotation_period = "100000s"
+}
+
+resource "google_storage_bucket" "temp" {
+  name = "%s"
+  force_destroy = true
+}
+
+resource "google_dataflow_job" "big_data" {
+  name = "%s"
+ 
+  zone    = "%s"
+
+  machine_type      = "e2-standard-2"
+  template_gcs_path = "%s"
+  temp_gcs_location = google_storage_bucket.temp.url
+  kms_key_name		= google_kms_crypto_key.crypto_key.self_link
+  parameters = {
+    inputFile = "%s"
+    output    = "${google_storage_bucket.temp.url}/output"
+  }
+  on_delete = "cancel"
+}
+`, key_ring, crypto_key, bucket, job, zone, testDataflowJobTemplateWordCountUrl, testDataflowJobSampleFileUrl)
+}
+
 func testAccDataflowJob_additionalExperiments(bucket string, job string, experiments []string) string {
 	return fmt.Sprintf(`
 resource "google_storage_bucket" "temp" {
diff --git a/website/docs/r/dataflow_job.html.markdown b/website/docs/r/dataflow_job.html.markdown
index 31c6b10c73..6ecd1d91b1 100644
--- a/website/docs/r/dataflow_job.html.markdown
+++ b/website/docs/r/dataflow_job.html.markdown
@@ -87,6 +87,7 @@ The following arguments are supported:
 * `network` - (Optional) The network to which VMs will be assigned. If it is not provided, "default" will be used.
 * `subnetwork` - (Optional) The subnetwork to which VMs will be assigned. Should be of the form "regions/REGION/subnetworks/SUBNETWORK".
 * `machine_type` - (Optional) The machine type to use for the job.
+* `kms_key_name` - (Optional) The name for the Cloud KMS key for the job. Key format is: `projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY`
 * `ip_configuration` - (Optional) The configuration for VM IPs.  Options are `"WORKER_IP_PUBLIC"` or `"WORKER_IP_PRIVATE"`.
 * `additional_experiments` - (Optional) List of experiments that should be used by the job. An example value is `["enable_stackdriver_agent_metrics"]`.