### Is there an existing issue for this?

- [X] I have searched the existing issues

### Feature description

Basically we want to only run the lambda when an insert is occurring in DynamoDB.
I tested using Dynobase deleting a record and the lambda still triggered.

```
resource "aws_lambda_event_source_mapping" "event-log" {
	event_source_arn  = aws_dynamodb_table.event-log.stream_arn
	function_name     = module.main.lambda-arn
	starting_position = "LATEST"
	enabled = true # change to false when truncating the table in localstack
	maximum_retry_attempts = 1 # limit retry on failure to 1 for now

	# https://dev.to/aws-builders/filtering-dynamodb-streams-before-lambda-p5p
	filter_criteria { # Doesn't seem to work on LocalStack for now
		filter {
			pattern = jsonencode({
				body = {
					eventName : ["INSERT"]
				}
			})
		}
	}
}
```


### 🧑‍💻 Implementation

_No response_

### Anything else?

 https://dev.to/aws-builders/filtering-dynamodb-streams-before-lambda-p5p
assert len(event["event"]["Records"]) == num_records_per_batch
                actual_record_ids = []
                for record in event["event"]["Records"]:
                    assert "eventID" in record
                    assert "eventSourceARN" in record
                    assert "eventSource" in record
                    assert "eventVersion" in record
                    assert "eventName" in record
                    assert "invokeIdentityArn" in record
                    assert "awsRegion" in record
                    assert "kinesis" in record
                    record_data = base64.b64decode(record["kinesis"]["data"]).decode("utf-8")
                    actual_record_id = json.loads(record_data)["record_id"]
                    actual_record_ids.append(actual_record_id)
                actual_record_ids.sort()
                assert actual_record_ids == [i for i in range(num_records_per_batch)]
            assert (
                invocation_events[1]["executionStart"] - invocation_events[0]["executionStart"]
            ) > 5
        finally:
            lambda_client.delete_event_source_mapping(UUID=uuid)
    def test_kinesis_event_source_trim_horizon(
        self,
        lambda_client,
        kinesis_client,
        create_lambda_function,
        kinesis_create_stream,
        wait_for_stream_ready,
        logs_client,
        lambda_su_role,