chore: update Kafka endpoint settings to include all attributes

.changelog/20904.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_dms_endpoint: Add `message_format`, `include_transaction_details`, `include_partition_value`, `partition_include_schema_table`, `include_table_alter_operations`, `include_control_details`, `message_max_bytes`, `include_null_and_empty`, `security_protocol`, `ssl_client_certificate_arn`, `ssl_client_key_arn`, `ssl_client_key_password`, `ssl_ca_certificate_arn`, `sasl_username`, `sasl_password` and `no_hex_prefix` arguments to `kafka_settings` configuration block
+```

aws/internal/service/dms/enum.go

@@ -1,5 +1,9 @@
 package dms
 
+const (
+	EndpointStatusDeleting = "deleting"
+)
+
 const (
 	EngineNameAurora                     = "aurora"
 	EngineNameAuroraPostgresql           = "aurora-postgresql"

aws/internal/service/dms/finder/finder.go

@@ -0,0 +1,43 @@
+package finder
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	dms "github.com/aws/aws-sdk-go/service/databasemigrationservice"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/tfresource"
+)
+
+func EndpointByID(conn *dms.DatabaseMigrationService, id string) (*dms.Endpoint, error) {
+	input := &dms.DescribeEndpointsInput{
+		Filters: []*dms.Filter{
+			{
+				Name:   aws.String("endpoint-id"),
+				Values: aws.StringSlice([]string{id}),
+			},
+		},
+	}
+
+	output, err := conn.DescribeEndpoints(input)
+
+	if tfawserr.ErrCodeEquals(err, dms.ErrCodeResourceNotFoundFault) {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || len(output.Endpoints) == 0 || output.Endpoints[0] == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	if count := len(output.Endpoints); count > 1 {
+		return nil, tfresource.NewTooManyResultsError(count, input)
+	}
+
+	return output.Endpoints[0], nil
+}

aws/internal/service/dms/waiter/status.go

@@ -0,0 +1,25 @@
+package waiter
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	dms "github.com/aws/aws-sdk-go/service/databasemigrationservice"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/service/dms/finder"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/tfresource"
+)
+
+func EndpointStatus(conn *dms.DatabaseMigrationService, id string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		output, err := finder.EndpointByID(conn, id)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		return output, aws.StringValue(output.Status), nil
+	}
+}

aws/internal/service/dms/waiter/waiter.go

@@ -0,0 +1,30 @@
+package waiter
+
+import (
+	"time"
+
+	dms "github.com/aws/aws-sdk-go/service/databasemigrationservice"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	tfdms "github.com/terraform-providers/terraform-provider-aws/aws/internal/service/dms"
+)
+
+const (
+	EndpointDeletedTimeout = 5 * time.Minute
+)
+
+func EndpointDeleted(conn *dms.DatabaseMigrationService, id string) (*dms.Endpoint, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{tfdms.EndpointStatusDeleting},
+		Target:  []string{},
+		Refresh: EndpointStatus(conn, id),
+		Timeout: EndpointDeletedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*dms.Endpoint); ok {
+		return output, err
+	}
+
+	return nil, err
+}

aws/provider_test.go

@@ -356,6 +356,7 @@ func testAccCheckResourceAttrRegionalReverseDnsService(resourceName, attributeNa
 	}
 }
 
+/*
 // testAccCheckResourceAttrHostnameWithPort ensures the Terraform state regexp matches a formatted DNS hostname with prefix, partition DNS suffix, and given port
 func testAccCheckResourceAttrHostnameWithPort(resourceName, attributeName, serviceName, hostnamePrefix string, port int) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
@@ -365,6 +366,7 @@ func testAccCheckResourceAttrHostnameWithPort(resourceName, attributeName, servi
 		return resource.TestCheckResourceAttr(resourceName, attributeName, hostname)(s)
 	}
 }
+*/
 
 // testAccCheckResourceAttrPrivateDnsName ensures the Terraform state exactly matches a private DNS name
 //