diff --git a/samtranslator/model/eventsources/pull.py b/samtranslator/model/eventsources/pull.py
index 537d9cd30..6af57a29d 100644
--- a/samtranslator/model/eventsources/pull.py
+++ b/samtranslator/model/eventsources/pull.py
@@ -110,6 +110,9 @@ def to_cloudformation(self, **kwargs):
 
         destination_config_policy = None
         if self.DestinationConfig:
+            if self.DestinationConfig.get("OnFailure") is None:
+                raise InvalidEventException(self.logical_id, "'OnFailure' is a required field for 'DestinationConfig'")
+
             # `Type` property is for sam to attach the right policies
             destination_type = self.DestinationConfig.get("OnFailure").get("Type")
 
@@ -120,10 +123,6 @@ def to_cloudformation(self, **kwargs):
                 # the values 'SQS' and 'SNS' are allowed. No intrinsics are allowed
                 if destination_type not in ["SQS", "SNS"]:
                     raise InvalidEventException(self.logical_id, "The only valid values for 'Type' are 'SQS' and 'SNS'")
-                if self.DestinationConfig.get("OnFailure") is None:
-                    raise InvalidEventException(
-                        self.logical_id, "'OnFailure' is a required field for " "'DestinationConfig'"
-                    )
                 if destination_type == "SQS":
                     queue_arn = self.DestinationConfig.get("OnFailure").get("Destination")
                     destination_config_policy = IAMRolePolicies().sqs_send_message_role_policy(
@@ -134,6 +133,7 @@ def to_cloudformation(self, **kwargs):
                     destination_config_policy = IAMRolePolicies().sns_publish_role_policy(
                         sns_topic_arn, self.logical_id
                     )
+
             lambda_eventsourcemapping.DestinationConfig = self.DestinationConfig
 
         if "role" in kwargs:
diff --git a/tests/translator/input/error_function_with_invalid_stream_eventsource_dest_type.yaml b/tests/translator/input/error_function_with_invalid_stream_eventsource_dest_type.yaml
new file mode 100644
index 000000000..ddbde8b65
--- /dev/null
+++ b/tests/translator/input/error_function_with_invalid_stream_eventsource_dest_type.yaml
@@ -0,0 +1,45 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+
+Parameters:
+  MyBatchingWindowParam:
+    Type: Number
+    Default: 45
+    Description: parameter for batching window in seconds
+
+Resources:
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: index.handler
+      InlineCode: |
+        exports.handler = async (event) => {
+            return {
+            statusCode: 200,
+            body: JSON.stringify(event),
+            headers: {}
+            }
+        }
+      Runtime: nodejs12.x
+      Policies:
+        - SQSSendMessagePolicy:
+            QueueName: !GetAtt MySqsQueue.QueueName
+      Events:
+        StreamEvent:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt KinesisStream.Arn
+            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
+            StartingPosition: LATEST
+            DestinationConfig:
+              OnFailure:
+                Type: INVALID_VALID
+                Destination: !Ref MySnsTopic
+
+  KinesisStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      ShardCount: 1
+
+  MySnsTopic:
+    Type: AWS::SNS::Topic
\ No newline at end of file
diff --git a/tests/translator/input/error_function_with_missing_on_failure_in_stream_event_destination_config.yaml b/tests/translator/input/error_function_with_missing_on_failure_in_stream_event_destination_config.yaml
new file mode 100644
index 000000000..67510e2c3
--- /dev/null
+++ b/tests/translator/input/error_function_with_missing_on_failure_in_stream_event_destination_config.yaml
@@ -0,0 +1,45 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+
+Parameters:
+  MyBatchingWindowParam:
+    Type: Number
+    Default: 45
+    Description: parameter for batching window in seconds
+
+Resources:
+  MyFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      Handler: index.handler
+      InlineCode: |
+        exports.handler = async (event) => {
+            return {
+            statusCode: 200,
+            body: JSON.stringify(event),
+            headers: {}
+            }
+        }
+      Runtime: nodejs12.x
+      Policies:
+        - SQSSendMessagePolicy:
+            QueueName: !GetAtt MySqsQueue.QueueName
+      Events:
+        StreamEvent:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt KinesisStream.Arn
+            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
+            StartingPosition: LATEST
+            DestinationConfig:
+              InvalidConfig:
+                Type: SNS
+                Destination: !Ref MySnsTopic
+
+  KinesisStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      ShardCount: 1
+
+  MySnsTopic:
+    Type: AWS::SNS::Topic
\ No newline at end of file
diff --git a/tests/translator/input/function_with_event_source_mapping.yaml b/tests/translator/input/function_with_event_source_mapping.yaml
index edea3e214..7a55afa02 100644
--- a/tests/translator/input/function_with_event_source_mapping.yaml
+++ b/tests/translator/input/function_with_event_source_mapping.yaml
@@ -59,6 +59,22 @@ Resources:
               OnFailure:
                 Type: SQS
                 Destination: !GetAtt MySqsQueue.Arn
+        StreamEventWithoutDestinationConfigType:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt KinesisStream1.Arn
+            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
+            StartingPosition: LATEST
+            DestinationConfig:
+              OnFailure:
+                Destination: !Ref MySnsTopic
+        StreamEventWithEmptyDestinationConfig:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt KinesisStream1.Arn
+            MaximumBatchingWindowInSeconds: !Ref MyBatchingWindowParam
+            StartingPosition: LATEST
+            DestinationConfig:
 
   KinesisStream:
     Type: AWS::Kinesis::Stream
diff --git a/tests/translator/output/aws-cn/error_function_with_invalid_stream_eventsource_dest_type.json b/tests/translator/output/aws-cn/error_function_with_invalid_stream_eventsource_dest_type.json
new file mode 100644
index 000000000..dc4f300bd
--- /dev/null
+++ b/tests/translator/output/aws-cn/error_function_with_invalid_stream_eventsource_dest_type.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/aws-cn/error_function_with_missing_on_failure_in_stream_event_destination_config.json b/tests/translator/output/aws-cn/error_function_with_missing_on_failure_in_stream_event_destination_config.json
new file mode 100644
index 000000000..dcc5d33fb
--- /dev/null
+++ b/tests/translator/output/aws-cn/error_function_with_missing_on_failure_in_stream_event_destination_config.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/aws-cn/function_with_event_source_mapping.json b/tests/translator/output/aws-cn/function_with_event_source_mapping.json
index c566b9490..0a042dd07 100644
--- a/tests/translator/output/aws-cn/function_with_event_source_mapping.json
+++ b/tests/translator/output/aws-cn/function_with_event_source_mapping.json
@@ -233,6 +233,49 @@
         }
       }
     },
+    "MyFunctionForBatchingExampleStreamEventWithoutDestinationConfigType": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST",
+        "DestinationConfig": {
+            "OnFailure": {
+                "Destination": {
+                    "Ref": "MySnsTopic"
+                }
+            }
+        }
+      }
+    },
+    "MyFunctionForBatchingExampleStreamEventWithEmptyDestinationConfig": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST"
+      }
+    },
     "KinesisStream": {
       "Type": "AWS::Kinesis::Stream",
       "Properties": {
diff --git a/tests/translator/output/aws-us-gov/error_function_with_invalid_stream_eventsource_dest_type.json b/tests/translator/output/aws-us-gov/error_function_with_invalid_stream_eventsource_dest_type.json
new file mode 100644
index 000000000..dc4f300bd
--- /dev/null
+++ b/tests/translator/output/aws-us-gov/error_function_with_invalid_stream_eventsource_dest_type.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/aws-us-gov/error_function_with_missing_on_failure_in_stream_event_destination_config.json b/tests/translator/output/aws-us-gov/error_function_with_missing_on_failure_in_stream_event_destination_config.json
new file mode 100644
index 000000000..dcc5d33fb
--- /dev/null
+++ b/tests/translator/output/aws-us-gov/error_function_with_missing_on_failure_in_stream_event_destination_config.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/aws-us-gov/function_with_event_source_mapping.json b/tests/translator/output/aws-us-gov/function_with_event_source_mapping.json
index a910aefed..554891d3f 100644
--- a/tests/translator/output/aws-us-gov/function_with_event_source_mapping.json
+++ b/tests/translator/output/aws-us-gov/function_with_event_source_mapping.json
@@ -233,6 +233,49 @@
         }
       }
     },
+    "MyFunctionForBatchingExampleStreamEventWithoutDestinationConfigType": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST",
+        "DestinationConfig": {
+            "OnFailure": {
+                "Destination": {
+                    "Ref": "MySnsTopic"
+                }
+            }
+        }
+      }
+    },
+    "MyFunctionForBatchingExampleStreamEventWithEmptyDestinationConfig": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST"
+      }
+    },
     "KinesisStream": {
       "Type": "AWS::Kinesis::Stream",
       "Properties": {
diff --git a/tests/translator/output/error_function_with_invalid_stream_eventsource_dest_type.json b/tests/translator/output/error_function_with_invalid_stream_eventsource_dest_type.json
new file mode 100644
index 000000000..dc4f300bd
--- /dev/null
+++ b/tests/translator/output/error_function_with_invalid_stream_eventsource_dest_type.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. The only valid values for 'Type' are 'SQS' and 'SNS'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/error_function_with_missing_on_failure_in_stream_event_destination_config.json b/tests/translator/output/error_function_with_missing_on_failure_in_stream_event_destination_config.json
new file mode 100644
index 000000000..dcc5d33fb
--- /dev/null
+++ b/tests/translator/output/error_function_with_missing_on_failure_in_stream_event_destination_config.json
@@ -0,0 +1,8 @@
+{
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+    }
+  ], 
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MyFunction] is invalid. Event with id [MyFunctionStreamEvent] is invalid. 'OnFailure' is a required field for 'DestinationConfig'"
+}
\ No newline at end of file
diff --git a/tests/translator/output/function_with_event_source_mapping.json b/tests/translator/output/function_with_event_source_mapping.json
index 2a2f668ef..fcff5cabc 100644
--- a/tests/translator/output/function_with_event_source_mapping.json
+++ b/tests/translator/output/function_with_event_source_mapping.json
@@ -233,6 +233,49 @@
         }
       }
     },
+    "MyFunctionForBatchingExampleStreamEventWithoutDestinationConfigType": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST",
+        "DestinationConfig": {
+            "OnFailure": {
+                "Destination": {
+                    "Ref": "MySnsTopic"
+                }
+            }
+        }
+      }
+    },
+    "MyFunctionForBatchingExampleStreamEventWithEmptyDestinationConfig": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "MaximumBatchingWindowInSeconds": {
+          "Ref": "MyBatchingWindowParam"
+        },
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream1",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "MyFunctionForBatchingExample"
+        },
+        "StartingPosition": "LATEST"
+      }
+    },
     "KinesisStream": {
       "Type": "AWS::Kinesis::Stream",
       "Properties": {