diff --git a/aws/resource_aws_kinesis_firehose_delivery_stream.go b/aws/resource_aws_kinesis_firehose_delivery_stream.go
index 5b7755a0ebb..5a9f9f4bd95 100644
--- a/aws/resource_aws_kinesis_firehose_delivery_stream.go
+++ b/aws/resource_aws_kinesis_firehose_delivery_stream.go
@@ -19,6 +19,14 @@ const (
 	firehoseDeliveryStreamStatusDeleted = "DESTROYED"
 )
 
+const (
+	firehoseDestinationTypeS3            = "s3"
+	firehoseDestinationTypeExtendedS3    = "extended_s3"
+	firehoseDestinationTypeElasticsearch = "elasticsearch"
+	firehoseDestinationTypeRedshift      = "redshift"
+	firehoseDestinationTypeSplunk        = "splunk"
+)
+
 func cloudWatchLoggingOptionsSchema() *schema.Schema {
 	return &schema.Schema{
 		Type:     schema.TypeList,
@@ -55,26 +63,30 @@ func s3ConfigurationSchema() *schema.Schema {
 		Elem: &schema.Resource{
 			Schema: map[string]*schema.Schema{
 				"bucket_arn": {
-					Type:     schema.TypeString,
-					Required: true,
+					Type:         schema.TypeString,
+					Required:     true,
+					ValidateFunc: validateArn,
 				},
 
 				"buffer_size": {
-					Type:     schema.TypeInt,
-					Optional: true,
-					Default:  5,
+					Type:         schema.TypeInt,
+					Optional:     true,
+					Default:      5,
+					ValidateFunc: validation.IntAtLeast(1),
 				},
 
 				"buffer_interval": {
-					Type:     schema.TypeInt,
-					Optional: true,
-					Default:  300,
+					Type:         schema.TypeInt,
+					Optional:     true,
+					Default:      300,
+					ValidateFunc: validation.IntAtLeast(60),
 				},
 
 				"compression_format": {
-					Type:     schema.TypeString,
-					Optional: true,
-					Default:  firehose.CompressionFormatUncompressed,
+					Type:         schema.TypeString,
+					Optional:     true,
+					Default:      firehose.CompressionFormatUncompressed,
+					ValidateFunc: validation.StringInSlice(firehose.CompressionFormat_Values(), false),
 				},
 
 				"kms_key_arn": {
@@ -84,8 +96,9 @@ func s3ConfigurationSchema() *schema.Schema {
 				},
 
 				"role_arn": {
-					Type:     schema.TypeString,
-					Required: true,
+					Type:         schema.TypeString,
+					Required:     true,
+					ValidateFunc: validateArn,
 				},
 
 				"prefix": {
@@ -122,15 +135,9 @@ func processingConfigurationSchema() *schema.Schema {
 								Elem: &schema.Resource{
 									Schema: map[string]*schema.Schema{
 										"parameter_name": {
-											Type:     schema.TypeString,
-											Required: true,
-											ValidateFunc: validation.StringInSlice([]string{
-												firehose.ProcessorParameterNameLambdaArn,
-												firehose.ProcessorParameterNameNumberOfRetries,
-												firehose.ProcessorParameterNameRoleArn,
-												firehose.ProcessorParameterNameBufferSizeInMbs,
-												firehose.ProcessorParameterNameBufferIntervalInSeconds,
-											}, false),
+											Type:         schema.TypeString,
+											Required:     true,
+											ValidateFunc: validation.StringInSlice(firehose.ProcessorParameterName_Values(), false),
 										},
 										"parameter_value": {
 											Type:         schema.TypeString,
@@ -141,11 +148,9 @@ func processingConfigurationSchema() *schema.Schema {
 								},
 							},
 							"type": {
-								Type:     schema.TypeString,
-								Required: true,
-								ValidateFunc: validation.StringInSlice([]string{
-									firehose.ProcessorTypeLambda,
-								}, false),
+								Type:         schema.TypeString,
+								Required:     true,
+								ValidateFunc: validation.StringInSlice(firehose.ProcessorType_Values(), false),
 							},
 						},
 					},
@@ -177,7 +182,6 @@ func flattenFirehoseElasticsearchConfiguration(description *firehose.Elasticsear
 
 	m := map[string]interface{}{
 		"cloudwatch_logging_options": flattenCloudwatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"domain_arn":                 aws.StringValue(description.DomainARN),
 		"role_arn":                   aws.StringValue(description.RoleARN),
 		"type_name":                  aws.StringValue(description.TypeName),
 		"index_name":                 aws.StringValue(description.IndexName),
@@ -187,6 +191,14 @@ func flattenFirehoseElasticsearchConfiguration(description *firehose.Elasticsear
 		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, aws.StringValue(description.RoleARN)),
 	}
 
+	if description.DomainARN != nil {
+		m["domain_arn"] = aws.StringValue(description.DomainARN)
+	}
+
+	if description.ClusterEndpoint != nil {
+		m["cluster_endpoint"] = aws.StringValue(description.ClusterEndpoint)
+	}
+
 	if description.BufferingHints != nil {
 		m["buffering_interval"] = int(aws.Int64Value(description.BufferingHints.IntervalInSeconds))
 		m["buffering_size"] = int(aws.Int64Value(description.BufferingHints.SizeInMBs))
@@ -624,7 +636,7 @@ func flattenKinesisFirehoseDeliveryStream(d *schema.ResourceData, s *firehose.De
 	if len(s.Destinations) > 0 {
 		destination := s.Destinations[0]
 		if destination.RedshiftDestinationDescription != nil {
-			d.Set("destination", "redshift")
+			d.Set("destination", firehoseDestinationTypeRedshift)
 			configuredPassword := d.Get("redshift_configuration.0.password").(string)
 			if err := d.Set("redshift_configuration", flattenFirehoseRedshiftConfiguration(destination.RedshiftDestinationDescription, configuredPassword)); err != nil {
 				return fmt.Errorf("error setting redshift_configuration: %s", err)
@@ -633,7 +645,7 @@ func flattenKinesisFirehoseDeliveryStream(d *schema.ResourceData, s *firehose.De
 				return fmt.Errorf("error setting s3_configuration: %s", err)
 			}
 		} else if destination.ElasticsearchDestinationDescription != nil {
-			d.Set("destination", "elasticsearch")
+			d.Set("destination", firehoseDestinationTypeElasticsearch)
 			if err := d.Set("elasticsearch_configuration", flattenFirehoseElasticsearchConfiguration(destination.ElasticsearchDestinationDescription)); err != nil {
 				return fmt.Errorf("error setting elasticsearch_configuration: %s", err)
 			}
@@ -641,20 +653,20 @@ func flattenKinesisFirehoseDeliveryStream(d *schema.ResourceData, s *firehose.De
 				return fmt.Errorf("error setting s3_configuration: %s", err)
 			}
 		} else if destination.SplunkDestinationDescription != nil {
-			d.Set("destination", "splunk")
+			d.Set("destination", firehoseDestinationTypeSplunk)
 			if err := d.Set("splunk_configuration", flattenFirehoseSplunkConfiguration(destination.SplunkDestinationDescription)); err != nil {
 				return fmt.Errorf("error setting splunk_configuration: %s", err)
 			}
 			if err := d.Set("s3_configuration", flattenFirehoseS3Configuration(destination.SplunkDestinationDescription.S3DestinationDescription)); err != nil {
 				return fmt.Errorf("error setting s3_configuration: %s", err)
 			}
-		} else if d.Get("destination").(string) == "s3" {
-			d.Set("destination", "s3")
+		} else if d.Get("destination").(string) == firehoseDestinationTypeS3 {
+			d.Set("destination", firehoseDestinationTypeS3)
 			if err := d.Set("s3_configuration", flattenFirehoseS3Configuration(destination.S3DestinationDescription)); err != nil {
 				return fmt.Errorf("error setting s3_configuration: %s", err)
 			}
 		} else {
-			d.Set("destination", "extended_s3")
+			d.Set("destination", firehoseDestinationTypeExtendedS3)
 			if err := d.Set("extended_s3_configuration", flattenFirehoseExtendedS3Configuration(destination.ExtendedS3DestinationDescription)); err != nil {
 				return fmt.Errorf("error setting extended_s3_configuration: %s", err)
 			}
@@ -752,11 +764,11 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 					return strings.ToLower(value)
 				},
 				ValidateFunc: validation.StringInSlice([]string{
-					"elasticsearch",
-					"extended_s3",
-					"redshift",
-					"s3",
-					"splunk",
+					firehoseDestinationTypeS3,
+					firehoseDestinationTypeExtendedS3,
+					firehoseDestinationTypeRedshift,
+					firehoseDestinationTypeElasticsearch,
+					firehoseDestinationTypeSplunk,
 				}, false),
 			},
 
@@ -770,8 +782,9 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"bucket_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:         schema.TypeString,
+							Required:     true,
+							ValidateFunc: validateArn,
 						},
 
 						"buffer_size": {
@@ -787,9 +800,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						},
 
 						"compression_format": {
-							Type:     schema.TypeString,
-							Optional: true,
-							Default:  firehose.CompressionFormatUncompressed,
+							Type:         schema.TypeString,
+							Optional:     true,
+							Default:      firehose.CompressionFormatUncompressed,
+							ValidateFunc: validation.StringInSlice(firehose.CompressionFormat_Values(), false),
 						},
 
 						"data_format_conversion_configuration": {
@@ -899,14 +913,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 																			Default:  0.05,
 																		},
 																		"compression": {
-																			Type:     schema.TypeString,
-																			Optional: true,
-																			Default:  firehose.OrcCompressionSnappy,
-																			ValidateFunc: validation.StringInSlice([]string{
-																				firehose.OrcCompressionNone,
-																				firehose.OrcCompressionSnappy,
-																				firehose.OrcCompressionZlib,
-																			}, false),
+																			Type:         schema.TypeString,
+																			Optional:     true,
+																			Default:      firehose.OrcCompressionSnappy,
+																			ValidateFunc: validation.StringInSlice(firehose.OrcCompression_Values(), false),
 																		},
 																		"dictionary_key_threshold": {
 																			Type:     schema.TypeFloat,
@@ -919,13 +929,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 																			Default:  false,
 																		},
 																		"format_version": {
-																			Type:     schema.TypeString,
-																			Optional: true,
-																			Default:  firehose.OrcFormatVersionV012,
-																			ValidateFunc: validation.StringInSlice([]string{
-																				firehose.OrcFormatVersionV011,
-																				firehose.OrcFormatVersionV012,
-																			}, false),
+																			Type:         schema.TypeString,
+																			Optional:     true,
+																			Default:      firehose.OrcFormatVersionV012,
+																			ValidateFunc: validation.StringInSlice(firehose.OrcFormatVersion_Values(), false),
 																		},
 																		"padding_tolerance": {
 																			Type:     schema.TypeFloat,
@@ -965,14 +972,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 																			ValidateFunc: validation.IntAtLeast(67108864),
 																		},
 																		"compression": {
-																			Type:     schema.TypeString,
-																			Optional: true,
-																			Default:  firehose.ParquetCompressionSnappy,
-																			ValidateFunc: validation.StringInSlice([]string{
-																				firehose.ParquetCompressionGzip,
-																				firehose.ParquetCompressionSnappy,
-																				firehose.ParquetCompressionUncompressed,
-																			}, false),
+																			Type:         schema.TypeString,
+																			Optional:     true,
+																			Default:      firehose.ParquetCompressionSnappy,
+																			ValidateFunc: validation.StringInSlice(firehose.ParquetCompression_Values(), false),
 																		},
 																		"enable_dictionary_compression": {
 																			Type:     schema.TypeBool,
@@ -993,13 +996,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 																			ValidateFunc: validation.IntAtLeast(65536),
 																		},
 																		"writer_version": {
-																			Type:     schema.TypeString,
-																			Optional: true,
-																			Default:  firehose.ParquetWriterVersionV1,
-																			ValidateFunc: validation.StringInSlice([]string{
-																				firehose.ParquetWriterVersionV1,
-																				firehose.ParquetWriterVersionV2,
-																			}, false),
+																			Type:         schema.TypeString,
+																			Optional:     true,
+																			Default:      firehose.ParquetWriterVersionV1,
+																			ValidateFunc: validation.StringInSlice(firehose.ParquetWriterVersion_Values(), false),
 																		},
 																	},
 																},
@@ -1031,8 +1031,9 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 													Computed: true,
 												},
 												"role_arn": {
-													Type:     schema.TypeString,
-													Required: true,
+													Type:         schema.TypeString,
+													Required:     true,
+													ValidateFunc: validateArn,
 												},
 												"table_name": {
 													Type:     schema.TypeString,
@@ -1062,8 +1063,9 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						},
 
 						"role_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:         schema.TypeString,
+							Required:     true,
+							ValidateFunc: validateArn,
 						},
 
 						"prefix": {
@@ -1072,13 +1074,10 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						},
 
 						"s3_backup_mode": {
-							Type:     schema.TypeString,
-							Optional: true,
-							Default:  firehose.S3BackupModeDisabled,
-							ValidateFunc: validation.StringInSlice([]string{
-								firehose.S3BackupModeDisabled,
-								firehose.S3BackupModeEnabled,
-							}, false),
+							Type:         schema.TypeString,
+							Optional:     true,
+							Default:      firehose.S3BackupModeDisabled,
+							ValidateFunc: validation.StringInSlice(firehose.S3BackupMode_Values(), false),
 						},
 
 						"s3_backup_configuration": s3ConfigurationSchema(),
@@ -1115,34 +1114,25 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						"processing_configuration": processingConfigurationSchema(),
 
 						"role_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:         schema.TypeString,
+							Required:     true,
+							ValidateFunc: validateArn,
 						},
 
 						"s3_backup_mode": {
-							Type:     schema.TypeString,
-							Optional: true,
-							Default:  firehose.S3BackupModeDisabled,
-							ValidateFunc: validation.StringInSlice([]string{
-								firehose.S3BackupModeDisabled,
-								firehose.S3BackupModeEnabled,
-							}, false),
+							Type:         schema.TypeString,
+							Optional:     true,
+							Default:      firehose.S3BackupModeDisabled,
+							ValidateFunc: validation.StringInSlice(firehose.S3BackupMode_Values(), false),
 						},
 
 						"s3_backup_configuration": s3ConfigurationSchema(),
 
 						"retry_duration": {
-							Type:     schema.TypeInt,
-							Optional: true,
-							Default:  3600,
-							ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
-								value := v.(int)
-								if value < 0 || value > 7200 {
-									errors = append(errors, fmt.Errorf(
-										"%q must be in the range from 0 to 7200 seconds.", k))
-								}
-								return
-							},
+							Type:         schema.TypeInt,
+							Optional:     true,
+							Default:      3600,
+							ValidateFunc: validation.IntBetween(0, 7200),
 						},
 
 						"copy_options": {
@@ -1172,36 +1162,24 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"buffering_interval": {
-							Type:     schema.TypeInt,
-							Optional: true,
-							Default:  300,
-							ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
-								value := v.(int)
-								if value < 60 || value > 900 {
-									errors = append(errors, fmt.Errorf(
-										"%q must be in the range from 60 to 900 seconds.", k))
-								}
-								return
-							},
+							Type:         schema.TypeInt,
+							Optional:     true,
+							Default:      300,
+							ValidateFunc: validation.IntBetween(60, 900),
 						},
 
 						"buffering_size": {
-							Type:     schema.TypeInt,
-							Optional: true,
-							Default:  5,
-							ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
-								value := v.(int)
-								if value < 1 || value > 100 {
-									errors = append(errors, fmt.Errorf(
-										"%q must be in the range from 1 to 100 MB.", k))
-								}
-								return
-							},
+							Type:         schema.TypeInt,
+							Optional:     true,
+							Default:      5,
+							ValidateFunc: validation.IntBetween(1, 100),
 						},
 
 						"domain_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:          schema.TypeString,
+							Optional:      true,
+							ValidateFunc:  validateArn,
+							ConflictsWith: []string{"elasticsearch_configuration.0.cluster_endpoint"},
 						},
 
 						"index_name": {
@@ -1210,59 +1188,37 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						},
 
 						"index_rotation_period": {
-							Type:     schema.TypeString,
-							Optional: true,
-							Default:  firehose.ElasticsearchIndexRotationPeriodOneDay,
-							ValidateFunc: validation.StringInSlice([]string{
-								firehose.ElasticsearchIndexRotationPeriodNoRotation,
-								firehose.ElasticsearchIndexRotationPeriodOneHour,
-								firehose.ElasticsearchIndexRotationPeriodOneDay,
-								firehose.ElasticsearchIndexRotationPeriodOneWeek,
-								firehose.ElasticsearchIndexRotationPeriodOneMonth,
-							}, false),
+							Type:         schema.TypeString,
+							Optional:     true,
+							Default:      firehose.ElasticsearchIndexRotationPeriodOneDay,
+							ValidateFunc: validation.StringInSlice(firehose.ElasticsearchIndexRotationPeriod_Values(), false),
 						},
 
 						"retry_duration": {
-							Type:     schema.TypeInt,
-							Optional: true,
-							Default:  300,
-							ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
-								value := v.(int)
-								if value < 0 || value > 7200 {
-									errors = append(errors, fmt.Errorf(
-										"%q must be in the range from 0 to 7200 seconds.", k))
-								}
-								return
-							},
+							Type:         schema.TypeInt,
+							Optional:     true,
+							Default:      300,
+							ValidateFunc: validation.IntBetween(0, 7200),
 						},
 
 						"role_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:         schema.TypeString,
+							Required:     true,
+							ValidateFunc: validateArn,
 						},
 
 						"s3_backup_mode": {
-							Type:     schema.TypeString,
-							ForceNew: true,
-							Optional: true,
-							Default:  firehose.ElasticsearchS3BackupModeFailedDocumentsOnly,
-							ValidateFunc: validation.StringInSlice([]string{
-								firehose.ElasticsearchS3BackupModeFailedDocumentsOnly,
-								firehose.ElasticsearchS3BackupModeAllDocuments,
-							}, false),
+							Type:         schema.TypeString,
+							ForceNew:     true,
+							Optional:     true,
+							Default:      firehose.ElasticsearchS3BackupModeFailedDocumentsOnly,
+							ValidateFunc: validation.StringInSlice(firehose.ElasticsearchS3BackupMode_Values(), false),
 						},
 
 						"type_name": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
-								value := v.(string)
-								if len(value) > 100 {
-									errors = append(errors, fmt.Errorf(
-										"%q cannot be longer than 100 characters", k))
-								}
-								return
-							},
+							Type:         schema.TypeString,
+							Optional:     true,
+							ValidateFunc: validation.StringLenBetween(0, 100),
 						},
 
 						"vpc_config": {
@@ -1301,6 +1257,11 @@ func resourceAwsKinesisFirehoseDeliveryStream() *schema.Resource {
 						"cloudwatch_logging_options": cloudWatchLoggingOptionsSchema(),
 
 						"processing_configuration": processingConfigurationSchema(),
+						"cluster_endpoint": {
+							Type:          schema.TypeString,
+							Optional:      true,
+							ConflictsWith: []string{"elasticsearch_configuration.0.domain_arn"},
+						},
 					},
 				},
 			},
@@ -1814,7 +1775,7 @@ func extractEncryptionConfiguration(s3 map[string]interface{}) *firehose.Encrypt
 	}
 
 	return &firehose.EncryptionConfiguration{
-		NoEncryptionConfig: aws.String("NoEncryption"),
+		NoEncryptionConfig: aws.String(firehose.NoEncryptionConfigNoEncryption),
 	}
 }
 
@@ -1941,7 +1902,6 @@ func createElasticsearchConfig(d *schema.ResourceData, s3Config *firehose.S3Dest
 
 	config := &firehose.ElasticsearchDestinationConfiguration{
 		BufferingHints:  extractBufferingHints(es),
-		DomainARN:       aws.String(es["domain_arn"].(string)),
 		IndexName:       aws.String(es["index_name"].(string)),
 		RetryOptions:    extractElasticSearchRetryOptions(es),
 		RoleARN:         aws.String(es["role_arn"].(string)),
@@ -1949,6 +1909,14 @@ func createElasticsearchConfig(d *schema.ResourceData, s3Config *firehose.S3Dest
 		S3Configuration: s3Config,
 	}
 
+	if v, ok := es["domain_arn"]; ok && v.(string) != "" {
+		config.DomainARN = aws.String(v.(string))
+	}
+
+	if v, ok := es["cluster_endpoint"]; ok && v.(string) != "" {
+		config.ClusterEndpoint = aws.String(v.(string))
+	}
+
 	if _, ok := es["cloudwatch_logging_options"]; ok {
 		config.CloudWatchLoggingOptions = extractCloudWatchLoggingConfiguration(es)
 	}
@@ -1982,7 +1950,6 @@ func updateElasticsearchConfig(d *schema.ResourceData, s3Update *firehose.S3Dest
 
 	update := &firehose.ElasticsearchDestinationUpdate{
 		BufferingHints: extractBufferingHints(es),
-		DomainARN:      aws.String(es["domain_arn"].(string)),
 		IndexName:      aws.String(es["index_name"].(string)),
 		RetryOptions:   extractElasticSearchRetryOptions(es),
 		RoleARN:        aws.String(es["role_arn"].(string)),
@@ -1990,6 +1957,14 @@ func updateElasticsearchConfig(d *schema.ResourceData, s3Update *firehose.S3Dest
 		S3Update:       s3Update,
 	}
 
+	if v, ok := es["domain_arn"]; ok && v.(string) != "" {
+		update.DomainARN = aws.String(v.(string))
+	}
+
+	if v, ok := es["cluster_endpoint"]; ok && v.(string) != "" {
+		update.ClusterEndpoint = aws.String(v.(string))
+	}
+
 	if _, ok := es["cloudwatch_logging_options"]; ok {
 		update.CloudWatchLoggingOptions = extractCloudWatchLoggingConfiguration(es)
 	}
@@ -2149,27 +2124,27 @@ func resourceAwsKinesisFirehoseDeliveryStreamCreate(d *schema.ResourceData, meta
 		createInput.DeliveryStreamType = aws.String(firehose.DeliveryStreamTypeDirectPut)
 	}
 
-	if d.Get("destination").(string) == "extended_s3" {
+	if d.Get("destination").(string) == firehoseDestinationTypeExtendedS3 {
 		extendedS3Config := createExtendedS3Config(d)
 		createInput.ExtendedS3DestinationConfiguration = extendedS3Config
 	} else {
 		s3Config := createS3Config(d)
 
-		if d.Get("destination").(string) == "s3" {
+		if d.Get("destination").(string) == firehoseDestinationTypeS3 {
 			createInput.S3DestinationConfiguration = s3Config
-		} else if d.Get("destination").(string) == "elasticsearch" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeElasticsearch {
 			esConfig, err := createElasticsearchConfig(d, s3Config)
 			if err != nil {
 				return err
 			}
 			createInput.ElasticsearchDestinationConfiguration = esConfig
-		} else if d.Get("destination").(string) == "redshift" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeRedshift {
 			rc, err := createRedshiftConfig(d, s3Config)
 			if err != nil {
 				return err
 			}
 			createInput.RedshiftDestinationConfiguration = rc
-		} else if d.Get("destination").(string) == "splunk" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeSplunk {
 			rc, err := createSplunkConfig(d, s3Config)
 			if err != nil {
 				return err
@@ -2243,7 +2218,7 @@ func validateAwsKinesisFirehoseSchema(d *schema.ResourceData) error {
 	_, s3Exists := d.GetOk("s3_configuration")
 	_, extendedS3Exists := d.GetOk("extended_s3_configuration")
 
-	if d.Get("destination").(string) == "extended_s3" {
+	if d.Get("destination").(string) == firehoseDestinationTypeExtendedS3 {
 		if !extendedS3Exists {
 			return fmt.Errorf(
 				"When destination is 'extended_s3', extended_s3_configuration is required",
@@ -2285,27 +2260,27 @@ func resourceAwsKinesisFirehoseDeliveryStreamUpdate(d *schema.ResourceData, meta
 		DestinationId:                  aws.String(d.Get("destination_id").(string)),
 	}
 
-	if d.Get("destination").(string) == "extended_s3" {
+	if d.Get("destination").(string) == firehoseDestinationTypeExtendedS3 {
 		extendedS3Config := updateExtendedS3Config(d)
 		updateInput.ExtendedS3DestinationUpdate = extendedS3Config
 	} else {
 		s3Config := updateS3Config(d)
 
-		if d.Get("destination").(string) == "s3" {
+		if d.Get("destination").(string) == firehoseDestinationTypeS3 {
 			updateInput.S3DestinationUpdate = s3Config
-		} else if d.Get("destination").(string) == "elasticsearch" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeElasticsearch {
 			esUpdate, err := updateElasticsearchConfig(d, s3Config)
 			if err != nil {
 				return err
 			}
 			updateInput.ElasticsearchDestinationUpdate = esUpdate
-		} else if d.Get("destination").(string) == "redshift" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeRedshift {
 			rc, err := updateRedshiftConfig(d, s3Config)
 			if err != nil {
 				return err
 			}
 			updateInput.RedshiftDestinationUpdate = rc
-		} else if d.Get("destination").(string) == "splunk" {
+		} else if d.Get("destination").(string) == firehoseDestinationTypeSplunk {
 			rc, err := updateSplunkConfig(d, s3Config)
 			if err != nil {
 				return err
diff --git a/aws/resource_aws_kinesis_firehose_delivery_stream_test.go b/aws/resource_aws_kinesis_firehose_delivery_stream_test.go
index f5a163989e1..cd4387d59a9 100644
--- a/aws/resource_aws_kinesis_firehose_delivery_stream_test.go
+++ b/aws/resource_aws_kinesis_firehose_delivery_stream_test.go
@@ -1050,10 +1050,73 @@ func TestAccAWSKinesisFirehoseDeliveryStream_ElasticsearchConfigUpdates(t *testi
 	policyName := fmt.Sprintf("tf_acc_policy_%s", rString)
 	roleName := fmt.Sprintf("tf_acc_role_%s", rString)
 	preConfig := fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchBasic,
-		ri, ri, ri, ri, ri)
+		ri, ri, ri, ri, ri, ri)
 	postConfig := testAccFirehoseAWSLambdaConfigBasic(funcName, policyName, roleName) +
 		fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchUpdate,
-			ri, ri, ri, ri, ri)
+			ri, ri, ri, ri, ri, ri)
+
+	updatedElasticSearchConfig := &firehose.ElasticsearchDestinationDescription{
+		BufferingHints: &firehose.ElasticsearchBufferingHints{
+			IntervalInSeconds: aws.Int64(500),
+		},
+		ProcessingConfiguration: &firehose.ProcessingConfiguration{
+			Enabled: aws.Bool(true),
+			Processors: []*firehose.Processor{
+				{
+					Type: aws.String("Lambda"),
+					Parameters: []*firehose.ProcessorParameter{
+						{
+							ParameterName:  aws.String("LambdaArn"),
+							ParameterValue: aws.String("valueNotTested"),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckKinesisFirehoseDeliveryStreamDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: preConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckKinesisFirehoseDeliveryStreamExists(resourceName, &stream),
+					testAccCheckAWSKinesisFirehoseDeliveryStreamAttributes(&stream, nil, nil, nil, nil, nil),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: postConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckKinesisFirehoseDeliveryStreamExists(resourceName, &stream),
+					testAccCheckAWSKinesisFirehoseDeliveryStreamAttributes(&stream, nil, nil, nil, updatedElasticSearchConfig, nil),
+				),
+			},
+		},
+	})
+}
+
+func TestAccAWSKinesisFirehoseDeliveryStream_ElasticsearchConfigEndpointUpdates(t *testing.T) {
+	var stream firehose.DeliveryStreamDescription
+
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+	ri := acctest.RandInt()
+	rString := acctest.RandString(8)
+	funcName := fmt.Sprintf("aws_kinesis_firehose_delivery_stream_test_%s", rString)
+	policyName := fmt.Sprintf("tf_acc_policy_%s", rString)
+	roleName := fmt.Sprintf("tf_acc_role_%s", rString)
+	preConfig := fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchEndpoint,
+		ri, ri, ri, ri, ri, ri)
+	postConfig := testAccFirehoseAWSLambdaConfigBasic(funcName, policyName, roleName) +
+		fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchEndpointUpdate,
+			ri, ri, ri, ri, ri, ri)
 
 	updatedElasticSearchConfig := &firehose.ElasticsearchDestinationDescription{
 		BufferingHints: &firehose.ElasticsearchBufferingHints{
@@ -1083,7 +1146,7 @@ func TestAccAWSKinesisFirehoseDeliveryStream_ElasticsearchConfigUpdates(t *testi
 			{
 				Config: preConfig,
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckKinesisFirehoseDeliveryStreamExists("aws_kinesis_firehose_delivery_stream.test", &stream),
+					testAccCheckKinesisFirehoseDeliveryStreamExists(resourceName, &stream),
 					testAccCheckAWSKinesisFirehoseDeliveryStreamAttributes(&stream, nil, nil, nil, nil, nil),
 				),
 			},
@@ -1095,7 +1158,7 @@ func TestAccAWSKinesisFirehoseDeliveryStream_ElasticsearchConfigUpdates(t *testi
 			{
 				Config: postConfig,
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckKinesisFirehoseDeliveryStreamExists("aws_kinesis_firehose_delivery_stream.test", &stream),
+					testAccCheckKinesisFirehoseDeliveryStreamExists(resourceName, &stream),
 					testAccCheckAWSKinesisFirehoseDeliveryStreamAttributes(&stream, nil, nil, nil, updatedElasticSearchConfig, nil),
 				),
 			},
@@ -1629,7 +1692,7 @@ data "aws_caller_identity" "current" {}
 data "aws_partition" "current" {}
 
 resource "aws_iam_role" "firehose" {
-  name = "tf_acctest_firehose_delivery_role_%d"
+  name = "tf-acc-test-%[1]d"
 
   assume_role_policy = <<EOF
 {
@@ -1654,7 +1717,7 @@ EOF
 }
 
 resource "aws_iam_role_policy" "firehose" {
-  name = "tf_acctest_firehose_delivery_policy_%d"
+  name = "tf-acc-test-%[1]d"
   role = aws_iam_role.firehose.id
 
   policy = <<EOF
@@ -1692,22 +1755,22 @@ EOF
 }
 
 resource "aws_s3_bucket" "bucket" {
-  bucket = "tf-test-bucket-%d"
+  bucket = "tf-test-bucket-%[1]d"
   acl    = "private"
 }
 
 resource "aws_cloudwatch_log_group" "test" {
-  name = "example-%d"
+  name = "tf-acc-test-%[1]d"
 }
 
 resource "aws_cloudwatch_log_stream" "test" {
-  name           = "sample-log-stream-test-%d"
+  name           = "tf-acc-test-%[1]d"
   log_group_name = aws_cloudwatch_log_group.test.name
 }
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   depends_on  = [aws_iam_role_policy.firehose]
-  name        = "terraform-kinesis-firehose-cloudwatch-%d"
+  name        = "terraform-kinesis-firehose-%[1]d"
   destination = "s3"
 
   s3_configuration {
@@ -1721,7 +1784,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
     }
   }
 }
-`, rInt, rInt, rInt, rInt, rInt, rInt)
+`, rInt)
 }
 
 var testAccKinesisFirehoseDeliveryStreamConfig_s3basic = testAccKinesisFirehoseDeliveryStreamBaseConfig + `
@@ -1883,12 +1946,12 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_DataFormatConversionConfiguration_Enabled(rName string, rInt int, enabled bool) string {
 	return fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamBaseConfig, rInt, rInt, rInt) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {
-  name = "%s"
+  name = %[1]q
 }
 
 resource "aws_glue_catalog_table" "test" {
   database_name = aws_glue_catalog_database.test.name
-  name          = "%s"
+  name          = %[1]q
 
   storage_descriptor {
     columns {
@@ -1900,7 +1963,7 @@ resource "aws_glue_catalog_table" "test" {
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   destination = "extended_s3"
-  name        = "%s"
+  name        = %[1]q
 
   extended_s3_configuration {
     bucket_arn = aws_s3_bucket.bucket.arn
@@ -1909,7 +1972,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
     role_arn    = aws_iam_role.firehose.arn
 
     data_format_conversion_configuration {
-      enabled = %t
+      enabled = %[2]t
 
       input_format_configuration {
         deserializer {
@@ -1933,18 +1996,18 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
   depends_on = [aws_iam_role_policy.firehose]
 }
-`, rName, rName, rName, enabled)
+`, rName, enabled)
 }
 
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_DataFormatConversionConfiguration_HiveJsonSerDe_Empty(rName string, rInt int) string {
 	return fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamBaseConfig, rInt, rInt, rInt) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {
-  name = "%s"
+  name = %[1]q
 }
 
 resource "aws_glue_catalog_table" "test" {
   database_name = aws_glue_catalog_database.test.name
-  name          = "%s"
+  name          = %[1]q
 
   storage_descriptor {
     columns {
@@ -1956,7 +2019,7 @@ resource "aws_glue_catalog_table" "test" {
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   destination = "extended_s3"
-  name        = "%s"
+  name        = %[1]q
 
   extended_s3_configuration {
     bucket_arn = aws_s3_bucket.bucket.arn
@@ -1987,7 +2050,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
   depends_on = [aws_iam_role_policy.firehose]
 }
-`, rName, rName, rName)
+`, rName)
 }
 
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_ExternalUpdate(rName string, rInt int) string {
@@ -2007,12 +2070,12 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_DataFormatConversionConfiguration_OpenXJsonSerDe_Empty(rName string, rInt int) string {
 	return fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamBaseConfig, rInt, rInt, rInt) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {
-  name = "%s"
+  name = %[1]q
 }
 
 resource "aws_glue_catalog_table" "test" {
   database_name = aws_glue_catalog_database.test.name
-  name          = "%s"
+  name          = %[1]q
 
   storage_descriptor {
     columns {
@@ -2024,7 +2087,7 @@ resource "aws_glue_catalog_table" "test" {
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   destination = "extended_s3"
-  name        = "%s"
+  name        = %[1]q
 
   extended_s3_configuration {
     bucket_arn = aws_s3_bucket.bucket.arn
@@ -2055,18 +2118,18 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
   depends_on = [aws_iam_role_policy.firehose]
 }
-`, rName, rName, rName)
+`, rName)
 }
 
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_DataFormatConversionConfiguration_OrcSerDe_Empty(rName string, rInt int) string {
 	return fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamBaseConfig, rInt, rInt, rInt) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {
-  name = "%s"
+  name = %[1]q
 }
 
 resource "aws_glue_catalog_table" "test" {
   database_name = aws_glue_catalog_database.test.name
-  name          = "%s"
+  name          = %[1]q
 
   storage_descriptor {
     columns {
@@ -2078,7 +2141,7 @@ resource "aws_glue_catalog_table" "test" {
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   destination = "extended_s3"
-  name        = "%s"
+  name        = %[1]q
 
   extended_s3_configuration {
     bucket_arn = aws_s3_bucket.bucket.arn
@@ -2109,18 +2172,18 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
   depends_on = [aws_iam_role_policy.firehose]
 }
-`, rName, rName, rName)
+`, rName)
 }
 
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_DataFormatConversionConfiguration_ParquetSerDe_Empty(rName string, rInt int) string {
 	return fmt.Sprintf(testAccKinesisFirehoseDeliveryStreamBaseConfig, rInt, rInt, rInt) + fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {
-  name = "%s"
+  name = %[1]q
 }
 
 resource "aws_glue_catalog_table" "test" {
   database_name = aws_glue_catalog_database.test.name
-  name          = "%s"
+  name          = %[1]q
 
   storage_descriptor {
     columns {
@@ -2132,7 +2195,7 @@ resource "aws_glue_catalog_table" "test" {
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   destination = "extended_s3"
-  name        = "%s"
+  name        = %[1]q
 
   extended_s3_configuration {
     bucket_arn = aws_s3_bucket.bucket.arn
@@ -2163,7 +2226,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
   depends_on = [aws_iam_role_policy.firehose]
 }
-`, rName, rName, rName)
+`, rName)
 }
 
 func testAccKinesisFirehoseDeliveryStreamConfig_ExtendedS3_ErrorOutputPrefix(rName string, rInt int, errorOutputPrefix string) string {
@@ -2494,6 +2557,11 @@ resource "aws_elasticsearch_domain" "test_cluster" {
     instance_type = "m4.large.elasticsearch"
   }
 
+  domain_endpoint_options {
+    enforce_https       = true
+    tls_security_policy = "Policy-Min-TLS-1-2-2019-07"
+  }
+
   ebs_options {
     ebs_enabled = true
     volume_size = 10
@@ -2501,9 +2569,8 @@ resource "aws_elasticsearch_domain" "test_cluster" {
 }
 
 resource "aws_iam_role_policy" "firehose-elasticsearch" {
-  name = "elasticsearch"
+  name = "tf-acc-test-%d"
   role = aws_iam_role.firehose.id
-
   policy = <<EOF
 {
   "Version": "2012-10-17",
@@ -2740,12 +2807,66 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 	}
   }`
 
+var testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchEndpoint = testAccKinesisFirehoseDeliveryStreamBaseElasticsearchConfig + `
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on = ["aws_iam_role_policy.firehose-elasticsearch"]
+
+  name        = "terraform-kinesis-firehose-es-%d"
+  destination = "elasticsearch"
+
+  s3_configuration {
+    role_arn   = aws_iam_role.firehose.arn
+    bucket_arn = aws_s3_bucket.bucket.arn
+  }
+
+  elasticsearch_configuration {
+    cluster_endpoint = "https://${aws_elasticsearch_domain.test_cluster.endpoint}"
+    role_arn         = aws_iam_role.firehose.arn
+    index_name       = "test"
+    type_name        = "test"
+  }
+}`
+
+var testAccKinesisFirehoseDeliveryStreamConfig_ElasticsearchEndpointUpdate = testAccKinesisFirehoseDeliveryStreamBaseElasticsearchConfig + `
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on = ["aws_iam_role_policy.firehose-elasticsearch"]
+
+  name        = "terraform-kinesis-firehose-es-%d"
+  destination = "elasticsearch"
+
+  s3_configuration {
+    role_arn   = aws_iam_role.firehose.arn
+    bucket_arn = aws_s3_bucket.bucket.arn
+  }
+
+  elasticsearch_configuration {
+    cluster_endpoint   = "https://${aws_elasticsearch_domain.test_cluster.endpoint}"
+    role_arn           = "${aws_iam_role.firehose.arn}"
+    index_name         = "test"
+    type_name          = "test"
+    buffering_interval = 500
+
+    processing_configuration {
+      enabled = false
+
+      processors {
+        type = "Lambda"
+
+        parameters {
+          parameter_name  = "LambdaArn"
+          parameter_value = "${aws_lambda_function.lambda_function_test.arn}:$LATEST"
+        }
+      }
+    }
+  }
+}`
+
 func testAccKinesisFirehoseDeliveryStreamConfig_missingProcessingConfiguration(rInt int) string {
 	return fmt.Sprintf(`
 data "aws_partition" "current" {}
 
 resource "aws_iam_role" "firehose" {
-  name = "tf_acctest_firehose_delivery_role_%d"
+  name = "tf_acctest_firehose_delivery_role_%[1]d"
 
   assume_role_policy = <<EOF
 {
@@ -2765,7 +2886,7 @@ EOF
 }
 
 resource "aws_iam_role_policy" "firehose" {
-  name = "tf_acctest_firehose_delivery_policy_%d"
+  name = "tf_acctest_firehose_delivery_policy_%[1]d"
   role = aws_iam_role.firehose.id
 
   policy = <<EOF
@@ -2803,12 +2924,12 @@ EOF
 }
 
 resource "aws_s3_bucket" "bucket" {
-  bucket = "tf-test-bucket-%d"
+  bucket = "tf-test-bucket-%[1]d"
   acl    = "private"
 }
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
-  name        = "terraform-kinesis-firehose-mpc-%d"
+  name        = "terraform-kinesis-firehose-mpc-%[1]d"
   destination = "extended_s3"
 
   extended_s3_configuration {
@@ -2820,5 +2941,5 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
     bucket_arn         = aws_s3_bucket.bucket.arn
   }
 }
-`, rInt, rInt, rInt, rInt)
+`, rInt)
 }
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 9f328479135..24cc7dc37b8 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -397,7 +397,8 @@ The `elasticsearch_configuration` object supports the following:
 
 * `buffering_interval` - (Optional) Buffer incoming data for the specified period of time, in seconds between 60 to 900, before delivering it to the destination.  The default value is 300s.
 * `buffering_size` - (Optional) Buffer incoming data to the specified size, in MBs between 1 to 100, before delivering it to the destination.  The default value is 5MB.
-* `domain_arn` - (Required) The ARN of the Amazon ES domain.  The IAM role must have permission for `DescribeElasticsearchDomain`, `DescribeElasticsearchDomains`, and `DescribeElasticsearchDomainConfig` after assuming `RoleARN`.  The pattern needs to be `arn:.*`.
+* `domain_arn` - (Optional) The ARN of the Amazon ES domain.  The IAM role must have permission for `DescribeElasticsearchDomain`, `DescribeElasticsearchDomains`, and `DescribeElasticsearchDomainConfig` after assuming `RoleARN`.  The pattern needs to be `arn:.*`. Conflicts with `cluster_endpoint`.
+* `cluster_endpoint` - (Optional) The endpoint to use when communicating with the cluster. Conflicts with `domain_arn`.
 * `index_name` - (Required) The Elasticsearch index name.
 * `index_rotation_period` - (Optional) The Elasticsearch index rotation period.  Index rotation appends a timestamp to the IndexName to facilitate expiration of old data.  Valid values are `NoRotation`, `OneHour`, `OneDay`, `OneWeek`, and `OneMonth`.  The default value is `OneDay`.
 * `retry_duration` - (Optional) After an initial failure to deliver to Amazon Elasticsearch, the total amount of time, in seconds between 0 to 7200, during which Firehose re-attempts delivery (including the first attempt).  After this time has elapsed, the failed documents are written to Amazon S3.  The default value is 300s.  There will be no retry if the value is 0.