Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add provisioned_poller_config for kafka in lambda event source … #40303

Merged
merged 11 commits into from
Nov 27, 2024
Merged

feat: add provisioned_poller_config for kafka in lambda event source … #40303

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
21a5c4b
feat: add provisioned_poller_config for kafka in lambda event source …
lvthillo Nov 25, 2024
e9b9008
feat: add changelog
lvthillo Nov 25, 2024
a025189
fix: fix formatting in doc
lvthillo Nov 25, 2024
8787d3a
fix: fix acctest
lvthillo Nov 26, 2024
51c8c4d
Tweak CHANGELOG entry.
ewbankkit Nov 27, 2024
868da5d
Tidy up.
ewbankkit Nov 27, 2024
ec8dab5
Merge branch 'main' into HEAD
ewbankkit Nov 27, 2024
3e1864c
Update website/docs/r/lambda_event_source_mapping.html.markdown
ewbankkit Nov 27, 2024
b1491cd
Update internal/service/lambda/event_source_mapping_test.go
ewbankkit Nov 27, 2024
955fd22
Fix markdown-lint 'MD009/no-trailing-spaces Trailing spaces [Expected…
ewbankkit Nov 27, 2024
d6119da
Tweak 'TestAccLambdaEventSourceMapping_selfManagedKafkaWithProvisione…
ewbankkit Nov 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion internal/service/lambda/event_source_mapping.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,7 +247,6 @@ func resourceEventSourceMapping() *schema.Resource {
"provisioned_poller_config": {
Type: schema.TypeList,
Optional: true,
Computed: true,
MaxItems: 1,
Elem: &schema.Resource{
Schema: map[string]*schema.Schema{
Expand Down Expand Up @@ -738,6 +737,15 @@ func resourceEventSourceMappingUpdate(ctx context.Context, d *schema.ResourceDat
input.ParallelizationFactor = aws.Int32(int32(d.Get("parallelization_factor").(int)))
}

if d.HasChange("provisioned_poller_config") {
if v, ok := d.GetOk("provisioned_poller_config"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
input.ProvisionedPollerConfig = expandProvisionedPollerConfig(v.([]interface{})[0].(map[string]interface{}))
} else {
// AWS ignores the removal if this is left as nil.
input.ProvisionedPollerConfig = &awstypes.ProvisionedPollerConfig{}
}
}

if d.HasChange("scaling_config") {
if v, ok := d.GetOk("scaling_config"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
input.ScalingConfig = expandScalingConfig(v.([]interface{})[0].(map[string]interface{}))
Expand Down
130 changes: 101 additions & 29 deletions internal/service/lambda/event_source_mapping_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ func TestAccLambdaEventSourceMapping_Kinesis_basic(t *testing.T) {
resource.TestCheckResourceAttr(resourceName, "function_response_types.#", "0"),
resource.TestCheckResourceAttr(resourceName, names.AttrKMSKeyARN, ""),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "0"),
resource.TestCheckResourceAttr(resourceName, "tumbling_window_in_seconds", "0"),
),
},
Expand Down Expand Up @@ -877,12 +878,11 @@ func TestAccLambdaEventSourceMapping_msk(t *testing.T) {
Config: testAccEventSourceMappingConfig_msk(rName, "100"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "batch_size", "100"),
resource.TestCheckResourceAttrPair(resourceName, "event_source_arn", eventSourceResourceName, names.AttrARN),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "60"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "0"),
resource.TestCheckResourceAttr(resourceName, "topics.#", "1"),
resource.TestCheckTypeSetElemAttr(resourceName, "topics.*", "test"),
),
Expand All @@ -904,12 +904,10 @@ func TestAccLambdaEventSourceMapping_msk(t *testing.T) {
Config: testAccEventSourceMappingConfig_msk(rName, "9999"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "batch_size", "9999"),
resource.TestCheckResourceAttrPair(resourceName, "event_source_arn", eventSourceResourceName, names.AttrARN),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "60"),
resource.TestCheckResourceAttr(resourceName, "topics.#", "1"),
resource.TestCheckTypeSetElemAttr(resourceName, "topics.*", "test"),
),
Expand Down Expand Up @@ -939,14 +937,12 @@ func TestAccLambdaEventSourceMapping_mskWithEventSourceConfig(t *testing.T) {
Config: testAccEventSourceMappingConfig_mskWithEventSourceConfig(rName, "100"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.0.consumer_group_id", "amazon-managed-test-group-id"),
resource.TestCheckResourceAttr(resourceName, "batch_size", "100"),
resource.TestCheckResourceAttrPair(resourceName, "event_source_arn", eventSourceResourceName, names.AttrARN),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "amazon_managed_kafka_event_source_config.0.consumer_group_id", "amazon-managed-test-group-id"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "80"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.minimum_pollers", "10"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "0"),
resource.TestCheckResourceAttr(resourceName, "topics.#", "1"),
resource.TestCheckTypeSetElemAttr(resourceName, "topics.*", "test"),
),
Expand Down Expand Up @@ -979,11 +975,11 @@ func TestAccLambdaEventSourceMapping_selfManagedKafka(t *testing.T) {
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "batch_size", "100"),
resource.TestCheckResourceAttr(resourceName, names.AttrEnabled, acctest.CtFalse),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
ewbankkit marked this conversation as resolved.
Show resolved Hide resolved
resource.TestCheckResourceAttr(resourceName, "self_managed_event_source.#", "1"),
resource.TestCheckResourceAttr(resourceName, "self_managed_event_source.0.endpoints.KAFKA_BOOTSTRAP_SERVERS", "test1:9092,test2:9092"),
resource.TestCheckResourceAttr(resourceName, "self_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "source_access_configuration.#", "3"),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "topics.#", "1"),
resource.TestCheckTypeSetElemAttr(resourceName, "topics.*", "test"),
),
Expand Down Expand Up @@ -1024,15 +1020,13 @@ func TestAccLambdaEventSourceMapping_selfManagedKafkaWithEventSourceConfig(t *te
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "batch_size", "100"),
resource.TestCheckResourceAttr(resourceName, names.AttrEnabled, acctest.CtFalse),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "0"),
resource.TestCheckResourceAttr(resourceName, "self_managed_event_source.#", "1"),
resource.TestCheckResourceAttr(resourceName, "self_managed_event_source.0.endpoints.KAFKA_BOOTSTRAP_SERVERS", "test1:9092,test2:9092"),
resource.TestCheckResourceAttr(resourceName, "self_managed_kafka_event_source_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "self_managed_kafka_event_source_config.0.consumer_group_id", "self-managed-test-group-id"),
resource.TestCheckResourceAttr(resourceName, "source_access_configuration.#", "3"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "80"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.minimum_pollers", "10"),
acctest.CheckResourceAttrRFC3339(resourceName, "last_modified"),
resource.TestCheckResourceAttr(resourceName, "topics.#", "1"),
resource.TestCheckTypeSetElemAttr(resourceName, "topics.*", "test"),
),
Expand All @@ -1047,6 +1041,53 @@ func TestAccLambdaEventSourceMapping_selfManagedKafkaWithEventSourceConfig(t *te
})
}

func TestAccLambdaEventSourceMapping_selfManagedKafkaWithProvisionedPollerConfig(t *testing.T) {
ctx := acctest.Context(t)
var v lambda.GetEventSourceMappingOutput
resourceName := "aws_lambda_event_source_mapping.test"
rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)

resource.ParallelTest(t, resource.TestCase{
PreCheck: func() { acctest.PreCheck(ctx, t) },
ErrorCheck: acctest.ErrorCheck(t, names.LambdaServiceID),
ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
CheckDestroy: testAccCheckEventSourceMappingDestroy(ctx),
Steps: []resource.TestStep{
{
Config: testAccEventSourceMappingConfig_selfManagedKafkaWithProvisionedPollerConfig(rName, "100", "test1:9092,test2:9092", "100", "null"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "100"),
ewbankkit marked this conversation as resolved.
Show resolved Hide resolved
resource.TestCheckResourceAttrSet(resourceName, "provisioned_poller_config.0.minimum_pollers"),
ewbankkit marked this conversation as resolved.
Show resolved Hide resolved
),
},
{
ResourceName: resourceName,
ImportState: true,
ImportStateVerify: true,
ImportStateVerifyIgnore: []string{"last_modified"},
},
{
Config: testAccEventSourceMappingConfig_selfManagedKafkaWithProvisionedPollerConfig(rName, "100", "test1:9092,test2:9092", "150", "15"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "1"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.maximum_pollers", "150"),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.0.minimum_pollers", "15"),
),
},
{
Config: testAccEventSourceMappingConfig_selfManagedKafka(rName, "100", "test1:9092,test2:9092"),
Check: resource.ComposeTestCheckFunc(
testAccCheckEventSourceMappingExists(ctx, resourceName, &v),
resource.TestCheckResourceAttr(resourceName, "provisioned_poller_config.#", "0"),
),
},
},
})
}

func TestAccLambdaEventSourceMapping_activeMQ(t *testing.T) {
ctx := acctest.Context(t)
if testing.Short() {
Expand Down Expand Up @@ -2377,10 +2418,6 @@ resource "aws_lambda_event_source_mapping" "test" {
topics = ["test"]
starting_position = "TRIM_HORIZON"

provisioned_poller_config {
maximum_pollers = 60
}

depends_on = [aws_iam_policy_attachment.test]
}
`, rName, batchSize))
Expand Down Expand Up @@ -2422,11 +2459,6 @@ resource "aws_lambda_event_source_mapping" "test" {
consumer_group_id = "amazon-managed-test-group-id"
}

provisioned_poller_config {
maximum_pollers = 80
minimum_pollers = 10
}

depends_on = [aws_iam_policy_attachment.test]
}
`, rName, batchSize))
Expand Down Expand Up @@ -2463,7 +2495,6 @@ resource "aws_lambda_event_source_mapping" "test" {
type = "VPC_SECURITY_GROUP"
uri = aws_security_group.test.id
}

}
`, rName, batchSize, kafkaBootstrapServers))
}
Expand Down Expand Up @@ -2503,13 +2534,54 @@ resource "aws_lambda_event_source_mapping" "test" {
type = "VPC_SECURITY_GROUP"
uri = aws_security_group.test.id
}
}
`, rName, batchSize, kafkaBootstrapServers))
}

func testAccEventSourceMappingConfig_selfManagedKafkaWithProvisionedPollerConfig(rName, batchSize, kafkaBootstrapServers, maxPollers, minPollers string) string {
if batchSize == "" {
batchSize = "null"
}
if maxPollers == "" {
maxPollers = "null"
}
if minPollers == "" {
minPollers = "null"
}

return acctest.ConfigCompose(testAccEventSourceMappingConfig_kafkaBase(rName), fmt.Sprintf(`
resource "aws_lambda_event_source_mapping" "test" {
batch_size = %[2]s
enabled = false
function_name = aws_lambda_function.test.arn
topics = ["test"]
starting_position = "TRIM_HORIZON"

self_managed_event_source {
endpoints = {
KAFKA_BOOTSTRAP_SERVERS = %[3]q
}
}

dynamic "source_access_configuration" {
for_each = aws_subnet.test[*].id
content {
type = "VPC_SUBNET"
uri = "subnet:${source_access_configuration.value}"
}
}

source_access_configuration {
type = "VPC_SECURITY_GROUP"
uri = aws_security_group.test.id
}

provisioned_poller_config {
maximum_pollers = 80
minimum_pollers = 10
maximum_pollers = %[4]s
minimum_pollers = %[5]s
}
}
`, rName, batchSize, kafkaBootstrapServers))
`, rName, batchSize, kafkaBootstrapServers, maxPollers, minPollers))
}

func testAccEventSourceMappingConfig_dynamoDBBatchSize(rName, batchSize string) string {
Expand Down