<!-- Love localstack? Please consider supporting our collective:
👉  https://opencollective.com/localstack/donate -->

# Type of request: This is a ...

[ ] bug report
[ x ] feature request

# Detailed description
I am trying to set up a localstack lambda function that has a locally-run kafka cluster as its event source. I am using aws cli v2. When I run the command to create event source mapping for my lambda function using aws cli, I get a "500 Internal Server Error".
...

## Expected behavior
A new lambda event source mapping should be created for my local kafka cluster.
...

## Actual behavior
Getting an Internal Server Error on aws cli output and getting a "KeyError: EventSourceArn" on localstack logs.
...

# Steps to reproduce
1. Created a lambda handler in python.
    ```python 3
    import json

    def handler(event, context):
        body = {
            "message": str(event)
        }
        response = {
            "statusCode": 200,
            "body": json.dumps(body)
        }
        return response

2. Created my lambda function.
    ```
    $  aws lambda create-function \
        --function-name fun1 \
        --handler lambda.handler \
        --runtime python3.8 \
        --role r1 \
        --zip-file fileb://l1.zip \
        --endpoint-url http://localhost:4566
    ```

3. Created a secret.
    ```
    aws secretsmanager create-secret --name localstack --endpoint-url http://localhost:4566
    ```

4. Created a kafka cluster and a topic called `t1` on my local system using the [official kafka binary](https://www.apache.org/dyn/closer.cgi?path=/kafka/2.7.0/kafka_2.13-2.7.0.tgz). It works without problems.

5. Tried to create event source mapping using the command mentioned [here](https://docs.aws.amazon.com/lambda/latest/dg/services-smaa-topic-add.html) in the official AWS website.
    ```
   $ aws lambda create-event-source-mapping \
      --topics t1 \
      --source-access-configuration Type=SASL_SCRAM_512_AUTH,URI=arn:aws:secretsmanager:us-east-1:000000000000:secret:localstack-TDIuI \
      --function-name arn:aws:lambda:us-east-1:000000000000:function:fun1 \
      --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["localhost:9092"]}}' \
      --endpoint-url http://localhost:4566
    ```
    
    Here, aws cli throws the following error message
    ```
    An error occurred (500) when calling the CreateEventSourceMapping operation (reached max retries: 2): <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
    <title>500 Internal Server Error</title>
    <h1>Internal Server Error</h1>
    <p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
    ```

    And the localstack log gives the following error message
    ```
    localstack_main | 2021-01-30T13:32:54:ERROR:lambda_api: Exception on /2015-03-31/event-source-mappings/ [POST]
    localstack_main | Traceback (most recent call last):
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
    localstack_main |     response = self.full_dispatch_request()
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
    localstack_main |     rv = self.handle_user_exception(e)
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask_cors/extension.py", line 165, in wrapped_function
    localstack_main |     return cors_after_request(app.make_response(f(*args, **kwargs)))
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
    localstack_main |     reraise(exc_type, exc_value, tb)
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
    localstack_main |     raise value
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
    localstack_main |     rv = self.dispatch_request()
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
    localstack_main |     return self.view_functions[rule.endpoint](**req.view_args)
    localstack_main |   File "/opt/code/localstack/localstack/services/awslambda/lambda_api.py", line 1457, in create_event_source_mapping
    localstack_main |     data['FunctionName'], data['EventSourceArn'], data.get('Enabled'), data.get('BatchSize')
    localstack_main | KeyError: 'EventSourceArn'
    ```

## Command used to start LocalStack
Copied the contents of docker-compose.yml from localstack repo as-is and ran the following command:
```
docker-compose up
```
...

## Client code (AWS SDK code snippet, or sequence of "awslocal" commands)
Mentioned above.
...

def prepare_java_opts(self, java_opts):
        lambda_executors.config.LAMBDA_JAVA_OPTS = java_opts
        result = lambda_executors.Util.get_java_opts()
        return result
    def test_get_java_lib_folder_classpath(self):
        jar_file = os.path.join(new_tmp_dir(), 'foo.jar')
        save_file(jar_file, '')
        classpath = lambda_executors.Util.get_java_classpath(jar_file)
        self.assertIn('.:foo.jar', classpath)
        self.assertIn('*.jar', classpath)
    def test_get_java_lib_folder_classpath_no_directories(self):
        base_dir = new_tmp_dir()
        jar_file = os.path.join(base_dir, 'foo.jar')
        save_file(jar_file, '')
        lib_file = os.path.join(base_dir, 'lib', 'lib.jar')
        mkdir(os.path.dirname(lib_file))
        save_file(lib_file, '')
        classpath = lambda_executors.Util.get_java_classpath(jar_file)
        self.assertIn(':foo.jar', classpath)
        self.assertIn('lib/lib.jar:', classpath)
        self.assertIn(':*.jar', classpath)
    def test_get_java_lib_folder_classpath_archive_is_None(self):
        self.assertRaises(TypeError, lambda_executors.Util.get_java_classpath, None)
    @mock.patch('localstack.services.awslambda.lambda_executors.store_cloudwatch_logs')
    def test_executor_store_logs_can_handle_milliseconds(self, mock_store_cloudwatch_logs):
        mock_details = mock.Mock()
        t_sec = time.time()  
        t_ms = time.time() * 1000  
        
        lambda_executors._store_logs(mock_details, 'mock log output', t_ms)