diff --git a/plugins/modules/ec2_vpc_endpoint.py b/plugins/modules/ec2_vpc_endpoint.py
index 8e2426a525a..28d0fda0eba 100644
--- a/plugins/modules/ec2_vpc_endpoint.py
+++ b/plugins/modules/ec2_vpc_endpoint.py
@@ -313,8 +313,16 @@ def setup_removal(client, module):
         params['VpcEndpointIds'] = module.params.get('vpc_endpoint_id')
     try:
         result = client.delete_vpc_endpoints(**params)['Unsuccessful']
-        if not module.check_mode and (result != []):
-            module.fail_json(msg=result)
+        if len(result) < len(params['VpcEndpointIds']):
+            changed = True
+        # For some reason delete_vpc_endpoints doesn't throw exceptions it
+        # returns a list of failed 'results' instead.  Throw these so we can
+        # catch them the way we expect
+        for r in result:
+            try:
+                raise botocore.exceptions.ClientError(r, 'delete_vpc_endpoints')
+            except is_boto3_error_code('InvalidVpcEndpoint.NotFound'):
+                continue
     except is_boto3_error_code('DryRunOperation'):
         changed = True
         result = 'Would have deleted VPC Endpoint if not in check mode'
diff --git a/tests/integration/targets/ec2_vpc_endpoint/aliases b/tests/integration/targets/ec2_vpc_endpoint/aliases
new file mode 100644
index 00000000000..0ed26a0a740
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_endpoint/aliases
@@ -0,0 +1,3 @@
+cloud/aws
+shippable/aws/group2
+ec2_vpc_endpoint_info
diff --git a/tests/integration/targets/ec2_vpc_endpoint/defaults/main.yml b/tests/integration/targets/ec2_vpc_endpoint/defaults/main.yml
new file mode 100644
index 00000000000..f0041c402e5
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_endpoint/defaults/main.yml
@@ -0,0 +1,8 @@
+---
+vpc_name: '{{ resource_prefix }}-vpc'
+vpc_seed: '{{ resource_prefix }}'
+vpc_cidr: '10.{{ 256 | random(seed=vpc_seed) }}.22.0/24'
+
+# S3 and EC2 should generally be available...
+endpoint_service_a: 'com.amazonaws.{{ aws_region }}.s3'
+endpoint_service_b: 'com.amazonaws.{{ aws_region }}.ec2'
diff --git a/tests/integration/targets/ec2_vpc_endpoint/tasks/main.yml b/tests/integration/targets/ec2_vpc_endpoint/tasks/main.yml
new file mode 100644
index 00000000000..d06ac92fa01
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_endpoint/tasks/main.yml
@@ -0,0 +1,395 @@
+---
+- name: ec2_vpc_endpoint tests
+  collections:
+    - amazon.aws
+
+  module_defaults:
+    group/aws:
+      aws_access_key: "{{ aws_access_key }}"
+      aws_secret_key: "{{ aws_secret_key }}"
+      security_token: "{{ security_token | default(omit) }}"
+      region: "{{ aws_region }}"
+  block:
+  # ============================================================
+  # BEGIN PRE-TEST SETUP
+  - name: create a VPC
+    ec2_vpc_net:
+      state: present
+      name: "{{ vpc_name }}"
+      cidr_block: "{{ vpc_cidr }}"
+      tags:
+        AnsibleTest: 'ec2_vpc_endpoint'
+        AnsibleRun: '{{ resource_prefix }}'
+    register: vpc_creation
+  - name: Assert success
+    assert:
+      that:
+        - vpc_creation is successful
+
+  - name: Create an IGW
+    ec2_vpc_igw:
+      vpc_id: "{{ vpc_creation.vpc.id }}"
+      state: present
+      tags:
+        Name: "{{ resource_prefix }}"
+        AnsibleTest: 'ec2_vpc_endpoint'
+        AnsibleRun: '{{ resource_prefix }}'
+    register: igw_creation
+  - name: Assert success
+    assert:
+      that:
+        - igw_creation is successful
+
+  - name: Create a minimal route table (no routes)
+    ec2_vpc_route_table:
+      vpc_id: '{{ vpc_creation.vpc.id }}'
+      tags:
+        AnsibleTest: 'ec2_vpc_endpoint'
+        AnsibleRun: '{{ resource_prefix }}'
+        Name: '{{ resource_prefix }}-empty'
+      subnets: []
+      routes: []
+    register: rtb_creation_empty
+
+  - name: Create a minimal route table (with IGW)
+    ec2_vpc_route_table:
+      vpc_id: '{{ vpc_creation.vpc.id }}'
+      tags:
+        AnsibleTest: 'ec2_vpc_endpoint'
+        AnsibleRun: '{{ resource_prefix }}'
+        Name: '{{ resource_prefix }}-igw'
+      subnets: []
+      routes:
+        - dest: 0.0.0.0/0
+          gateway_id: "{{ igw_creation.gateway_id }}"
+    register: rtb_creation_igw
+
+  - name: Save VPC info in a fact
+    set_fact:
+      vpc_id: '{{ vpc_creation.vpc.id }}'
+      rtb_empty_id: '{{ rtb_creation_empty.route_table.id }}'
+      rtb_igw_id: '{{ rtb_creation_igw.route_table.id }}'
+
+  # ============================================================
+  # BEGIN TESTS
+
+  # Minimal check_mode with _info
+  - name: Fetch Endpoints in check_mode
+    ec2_vpc_endpoint_info:
+      query: endpoints
+    register: endpoint_info
+    check_mode: True
+  - name: Assert success
+    assert:
+      that:
+        # May be run in parallel, the only thing we can guarantee is
+        # - we shouldn't error
+        # - we should return 'vpc_endpoints' (even if it's empty)
+        - endpoint_info is successful
+        - '"vpc_endpoints" in endpoint_info'
+
+  - name: Fetch Services in check_mode
+    ec2_vpc_endpoint_info:
+      query: services
+    register: endpoint_info
+    check_mode: True
+  - name: Assert success
+    assert:
+      that:
+        - endpoint_info is successful
+        - '"service_names" in endpoint_info'
+        # This is just 2 arbitrary AWS services that should (generally) be
+        # available.  The actual list will vary over time and between regions
+        - endpoint_service_a in endpoint_info.service_names
+        - endpoint_service_b in endpoint_info.service_names
+
+  # Fetch services without check mode
+  # Note: Filters not supported on services via this module, this is all we can test for now
+  - name: Fetch Services
+    ec2_vpc_endpoint_info:
+      query: services
+    register: endpoint_info
+  - name: Assert success
+    assert:
+      that:
+        - endpoint_info is successful
+        - '"service_names" in endpoint_info'
+        # This is just 2 arbitrary AWS services that should (generally) be
+        # available.  The actual list will vary over time and between regions
+        - endpoint_service_a in endpoint_info.service_names
+        - endpoint_service_b in endpoint_info.service_names
+
+  # Attempt to create an endpoint
+  - name: Create minimal endpoint (check mode)
+    ec2_vpc_endpoint:
+      state: present
+      vpc_id: '{{ vpc_id }}'
+      service: '{{ endpoint_service_a }}'
+    register: create_endpoint_check
+    check_mode: True
+  - name: Assert changed
+    assert:
+      that:
+        - create_endpoint_check is changed
+
+  - name: Create minimal endpoint
+    ec2_vpc_endpoint:
+      state: present
+      vpc_id: '{{ vpc_id }}'
+      service: '{{ endpoint_service_a }}'
+    register: create_endpoint
+  - name: Check standard return values
+    assert:
+      that:
+        - create_endpoint is changed
+        - '"result" in create_endpoint'
+        - '"creation_timestamp" in create_endpoint.result'
+        - '"dns_entries" in create_endpoint.result'
+        - '"groups" in create_endpoint.result'
+        - '"network_interface_ids" in create_endpoint.result'
+        - '"owner_id" in create_endpoint.result'
+        - '"policy_document" in create_endpoint.result'
+        - '"private_dns_enabled" in create_endpoint.result'
+        - create_endpoint.result.private_dns_enabled == False
+        - '"requester_managed" in create_endpoint.result'
+        - create_endpoint.result.requester_managed == False
+        - '"service_name" in create_endpoint.result'
+        - create_endpoint.result.service_name == endpoint_service_a
+        - '"state" in create_endpoint.result'
+        - create_endpoint.result.state == "available"
+        - '"vpc_endpoint_id" in create_endpoint.result'
+        - create_endpoint.result.vpc_endpoint_id.startswith("vpce-")
+        - '"vpc_endpoint_type" in create_endpoint.result'
+        - create_endpoint.result.vpc_endpoint_type == "Gateway"
+        - '"vpc_id" in create_endpoint.result'
+        - create_endpoint.result.vpc_id == vpc_id
+
+  - name: Save Endpoint info in a fact
+    set_fact:
+      endpoint_id: '{{ create_endpoint.result.vpc_endpoint_id }}'
+
+  # Pull info about the endpoints
+  - name: Fetch Endpoints (all)
+    ec2_vpc_endpoint_info:
+      query: endpoints
+    register: endpoint_info
+  - name: Assert success
+    assert:
+      that:
+        # We're fetching all endpoints, there's no guarantee what the values
+        # will be
+        - endpoint_info is successful
+        - '"vpc_endpoints" in endpoint_info'
+        - '"creation_timestamp" in first_endpoint'
+        - '"policy_document" in first_endpoint'
+        - '"route_table_ids" in first_endpoint'
+        - '"service_name" in first_endpoint'
+        - '"state" in first_endpoint'
+        - '"vpc_endpoint_id" in first_endpoint'
+        - '"vpc_id" in first_endpoint'
+        # Not yet documented, but returned
+        - '"dns_entries" in first_endpoint'
+        - '"groups" in first_endpoint'
+        - '"network_interface_ids" in first_endpoint'
+        - '"owner_id" in first_endpoint'
+        - '"private_dns_enabled" in first_endpoint'
+        - '"requester_managed" in first_endpoint'
+        - '"subnet_ids" in first_endpoint'
+        - '"tags" in first_endpoint'
+        - '"vpc_endpoint_type" in first_endpoint'
+        # Make sure our endpoint is included
+        - endpoint_id in ( endpoint_info | community.general.json_query("vpc_endpoints[*].vpc_endpoint_id") | list | flatten )
+    vars:
+      first_endpoint: '{{ endpoint_info.vpc_endpoints[0] }}'
+
+  - name: Fetch Endpoints (targetted by ID)
+    ec2_vpc_endpoint_info:
+      query: endpoints
+      vpc_endpoint_ids: '{{ endpoint_id }}'
+    register: endpoint_info
+  - name: Assert success
+    assert:
+      that:
+        - endpoint_info is successful
+        - '"vpc_endpoints" in endpoint_info'
+        - '"creation_timestamp" in first_endpoint'
+        - '"policy_document" in first_endpoint'
+        - '"route_table_ids" in first_endpoint'
+        - '"service_name" in first_endpoint'
+        - first_endpoint.service_name == endpoint_service_a
+        - '"state" in first_endpoint'
+        - first_endpoint.state == "available"
+        - '"vpc_endpoint_id" in first_endpoint'
+        - first_endpoint.vpc_endpoint_id == endpoint_id
+        - '"vpc_id" in first_endpoint'
+        - first_endpoint.vpc_id == vpc_id
+        # Not yet documented, but returned
+        - '"dns_entries" in first_endpoint'
+        - '"groups" in first_endpoint'
+        - '"network_interface_ids" in first_endpoint'
+        - '"owner_id" in first_endpoint'
+        - '"private_dns_enabled" in first_endpoint'
+        - first_endpoint.private_dns_enabled == False
+        - '"requester_managed" in first_endpoint'
+        - first_endpoint.requester_managed == False
+        - '"subnet_ids" in first_endpoint'
+        - '"tags" in first_endpoint'
+        - '"vpc_endpoint_type" in first_endpoint'
+    vars:
+      first_endpoint: '{{ endpoint_info.vpc_endpoints[0] }}'
+
+  - name: Fetch Endpoints (targetted by VPC)
+    ec2_vpc_endpoint_info:
+      query: endpoints
+      filters:
+        vpc-id:
+        - '{{ vpc_id }}'
+    register: endpoint_info
+  - name: Assert success
+    assert:
+      that:
+        - endpoint_info is successful
+        - '"vpc_endpoints" in endpoint_info'
+        - '"creation_timestamp" in first_endpoint'
+        - '"policy_document" in first_endpoint'
+        - '"route_table_ids" in first_endpoint'
+        - '"service_name" in first_endpoint'
+        - first_endpoint.service_name == endpoint_service_a
+        - '"state" in first_endpoint'
+        - first_endpoint.state == "available"
+        - '"vpc_endpoint_id" in first_endpoint'
+        - first_endpoint.vpc_endpoint_id == endpoint_id
+        - '"vpc_id" in first_endpoint'
+        - first_endpoint.vpc_id == vpc_id
+        # Not yet documented, but returned
+        - '"dns_entries" in first_endpoint'
+        - '"groups" in first_endpoint'
+        - '"network_interface_ids" in first_endpoint'
+        - '"owner_id" in first_endpoint'
+        - '"private_dns_enabled" in first_endpoint'
+        - first_endpoint.private_dns_enabled == False
+        - '"requester_managed" in first_endpoint'
+        - first_endpoint.requester_managed == False
+        - '"subnet_ids" in first_endpoint'
+        - '"tags" in first_endpoint'
+        - '"vpc_endpoint_type" in first_endpoint'
+    vars:
+      first_endpoint: '{{ endpoint_info.vpc_endpoints[0] }}'
+
+# ec2_vpc_endpoint is not idempotent without explicitly passing the endpoint ID
+#  - name: Create minimal endpoint - idempotency (check mode)
+#    ec2_vpc_endpoint:
+#      state: present
+#      vpc_id: '{{ vpc_id }}'
+#      service: '{{ endpoint_service_a }}'
+#    register: create_endpoint_idem_check
+#    check_mode: True
+#
+#  - name: Create minimal endpoint - idempotency
+#    ec2_vpc_endpoint:
+#      state: present
+#      vpc_id: '{{ vpc_id }}'
+#      service: '{{ endpoint_service_a }}'
+#    register: create_endpoint_idem
+
+  # Deletion
+  # Delete the routes first - you can't delete an endpoint with a route
+  # attached.
+  - name: Delete minimal route table (no routes)
+    ec2_vpc_route_table:
+      state: absent
+      lookup: id
+      route_table_id: "{{ rtb_empty_id }}"
+    register: rtb_delete
+  - assert:
+      that:
+        - rtb_delete is changed
+
+  - name: Delete minimal route table (IGW route)
+    ec2_vpc_route_table:
+      state: absent
+      lookup: id
+      route_table_id: "{{ rtb_igw_id }}"
+  - assert:
+      that:
+        - rtb_delete is changed
+
+  - name: Delete endpoint by ID (check_mode)
+    ec2_vpc_endpoint:
+      state: absent
+      vpc_endpoint_id: "{{ endpoint_id }}"
+    check_mode: true
+    register: endpoint_delete_check
+  - assert:
+      that:
+        - endpoint_delete_check is changed
+
+  - name: Delete endpoint by ID
+    ec2_vpc_endpoint:
+      state: absent
+      vpc_endpoint_id: "{{ endpoint_id }}"
+    register: endpoint_delete_check
+  - assert:
+      that:
+        - endpoint_delete_check is changed
+
+# XXX Bug: uses AWS's check mode which only checks permissions
+#  - name: Delete endpoint by ID - idempotency (check_mode)
+#    ec2_vpc_endpoint:
+#      state: absent
+#      vpc_endpoint_id: "{{ endpoint_id }}"
+#    check_mode: true
+#    register: endpoint_delete_check
+#  - assert:
+#      that:
+#        - endpoint_delete_check is not changed
+
+  - name: Delete endpoint by ID - idempotency
+    ec2_vpc_endpoint:
+      state: absent
+      vpc_endpoint_id: "{{ endpoint_id }}"
+    register: endpoint_delete_check
+  - assert:
+      that:
+        - endpoint_delete_check is not changed
+
+
+  # ============================================================
+  # BEGIN POST-TEST CLEANUP
+  always:
+  - name: Delete minimal route table (no routes)
+    ec2_vpc_route_table:
+      state: absent
+      lookup: id
+      route_table_id: "{{ rtb_creation_empty.route_table.id }}"
+    ignore_errors: True
+
+  - name: Delete minimal route table (IGW route)
+    ec2_vpc_route_table:
+      state: absent
+      lookup: id
+      route_table_id: "{{ rtb_creation_igw.route_table.id }}"
+    ignore_errors: True
+
+  - name: Delete endpoint
+    ec2_vpc_endpoint:
+      state: absent
+      vpc_endpoint_id: "{{ create_endpoint.result.vpc_endpoint_id }}"
+    ignore_errors: True
+
+  - name: Remove IGW
+    ec2_vpc_igw:
+      state: absent
+      vpc_id: "{{ vpc_creation.vpc.id }}"
+    register: igw_deletion
+    retries: 10
+    delay: 5
+    until: igw_deletion is success
+    ignore_errors: yes
+
+  - name: Remove VPC
+    ec2_vpc_net:
+      state: absent
+      name: "{{ vpc_name }}"
+      cidr_block: "{{ vpc_cidr }}"
+    ignore_errors: true