Fix:delete corresponding event source mappings on deletion of dynamod…
diff --git a/localstack/services/dynamodb/dynamodb_listener.py b/localstack/services/dynamodb/dynamodb_listener.py
index c56ded01a399f..8064e9e7d6042 100644
--- a/localstack/services/dynamodb/dynamodb_listener.py
+++ b/localstack/services/dynamodb/dynamodb_listener.py
@@ -277,12 +277,13 @@ def return_response(self, method, path, data, headers, response):
             return
 
         elif action == '%s.DeleteTable' % ACTION_PREFIX:
+            table_arn = json.loads(response._content).get('TableDescription', {}).get('TableArn')
             event_publisher.fire_event(
                 event_publisher.EVENT_DYNAMODB_DELETE_TABLE,
                 payload={'n': event_publisher.get_hash(data['TableName'])}
             )
-
-            TABLE_TAGS.pop(json.loads(response._content).get('TableDescription', {}).get('TableArn', None), None)
+            self.delete_all_event_source_mappings(table_arn)
+            TABLE_TAGS.pop(table_arn, None)
 
             return
 
@@ -401,6 +402,13 @@ def prepare_transact_write_item_records(self, record, data):
                 records.append(new_record)
         return records
 
+    def delete_all_event_source_mappings(self, table_arn):
+        lambda_client = aws_stack.connect_to_service('lambda')
+        result = lambda_client.list_event_source_mappings(EventSourceArn=table_arn)
+        for event in result['EventSourceMappings']:
+            event_source_mapping_id = event['UUID']
+            lambda_client.delete_event_source_mapping(UUID=event_source_mapping_id)
+
     @staticmethod
     def _thread_local(name, default=None):
         try:
diff --git a/tests/integration/test_lambda.py b/tests/integration/test_lambda.py
index e7dc1fad11ef1..d9bc999cba7f6 100644
--- a/tests/integration/test_lambda.py
+++ b/tests/integration/test_lambda.py
@@ -370,6 +370,32 @@ def test_disabled_event_source_mapping_with_dynamodb(self):
 
         lambda_client.delete_function(FunctionName=function_name)
 
+    def test_deletion_event_source_mapping_with_dynamodb(self):
+        function_name = 'lambda_func-{}'.format(short_uid())
+        ddb_table = 'ddb_table-{}'.format(short_uid())
+
+        testutil.create_lambda_function(
+            handler_file=TEST_LAMBDA_ECHO_FILE,
+            func_name=function_name,
+            runtime=LAMBDA_RUNTIME_PYTHON36
+        )
+
+        table_arn = aws_stack.create_dynamodb_table(ddb_table, partition_key='id')['TableDescription']['TableArn']
+        lambda_client = aws_stack.connect_to_service('lambda')
+
+        lambda_client.create_event_source_mapping(
+            FunctionName=function_name,
+            EventSourceArn=table_arn
+        )
+
+        dynamodb_client = aws_stack.connect_to_service('dynamodb')
+        dynamodb_client.delete_table(TableName=ddb_table)
+
+        result = lambda_client.list_event_source_mappings(EventSourceArn=table_arn)
+        self.assertEqual(len(result['EventSourceMappings']), 0)
+        # clean up
+        lambda_client.delete_function(FunctionName=function_name)
+
 
 class TestPythonRuntimes(LambdaTestBase):
     @classmethod

Fix:delete corresponding event source mappings on deletion of dynamodb table.
#2561 