diff --git a/changelogs/fragments/sns_topic_type.yml b/changelogs/fragments/sns_topic_type.yml
new file mode 100644
index 00000000000..3c57bb3be36
--- /dev/null
+++ b/changelogs/fragments/sns_topic_type.yml
@@ -0,0 +1,2 @@
+minor_changes:
+  - sns_topic - Added ``topic_type`` parameter to select type of SNS topic (either FIFO or Standard) (https://github.com/ansible-collections/community.aws/pull/599).
diff --git a/plugins/modules/sns_topic.py b/plugins/modules/sns_topic.py
index 10d98e3034d..dd5af417bab 100644
--- a/plugins/modules/sns_topic.py
+++ b/plugins/modules/sns_topic.py
@@ -24,6 +24,13 @@
       - The name or ARN of the SNS topic to manage.
     required: true
     type: str
+  topic_type:
+    description:
+      - The type of topic that should be created. Either Standard for FIFO (first-in, first-out)
+    choices: ['standard', 'fifo']
+    default: 'standard'
+    type: str
+    version_added: 2.0.0
   state:
     description:
       - Whether to create or destroy an SNS topic.
@@ -228,6 +235,7 @@ class SnsTopicManager(object):
     def __init__(self,
                  module,
                  name,
+                 topic_type,
                  state,
                  display_name,
                  policy,
@@ -239,6 +247,7 @@ def __init__(self,
         self.connection = module.client('sns')
         self.module = module
         self.name = name
+        self.topic_type = topic_type
         self.state = state
         self.display_name = display_name
         self.policy = policy
@@ -285,9 +294,17 @@ def _topic_arn_lookup(self):
                 return topic
 
     def _create_topic(self):
+        attributes = {'FifoTopic': 'false'}
+        tags = []
+
+        if self.topic_type == 'fifo':
+            attributes['FifoTopic'] = 'true'
+            if not self.name.endswith('.fifo'):
+                self.name = self.name + '.fifo'
+
         if not self.check_mode:
             try:
-                response = self.connection.create_topic(Name=self.name)
+                response = self.connection.create_topic(Name=self.name, Attributes=attributes, Tags=tags)
             except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
                 self.module.fail_json_aws(e, msg="Couldn't create topic %s" % self.name)
             self.topic_arn = response['TopicArn']
@@ -456,6 +473,7 @@ def ensure_gone(self):
     def get_info(self):
         info = {
             'name': self.name,
+            'topic_type': self.topic_type,
             'state': self.state,
             'subscriptions_new': self.subscriptions,
             'subscriptions_existing': self.subscriptions_existing,
@@ -479,6 +497,7 @@ def get_info(self):
 def main():
     argument_spec = dict(
         name=dict(required=True),
+        topic_type=dict(type='str', default='standard', choices=['standard', 'fifo']),
         state=dict(default='present', choices=['present', 'absent']),
         display_name=dict(),
         policy=dict(type='dict'),
@@ -491,6 +510,7 @@ def main():
                               supports_check_mode=True)
 
     name = module.params.get('name')
+    topic_type = module.params.get('topic_type')
     state = module.params.get('state')
     display_name = module.params.get('display_name')
     policy = module.params.get('policy')
@@ -501,6 +521,7 @@ def main():
 
     sns_topic = SnsTopicManager(module,
                                 name,
+                                topic_type,
                                 state,
                                 display_name,
                                 policy,
diff --git a/tests/integration/targets/sns_topic/aliases b/tests/integration/targets/sns_topic/aliases
index e4280272565..4ef4b2067d0 100644
--- a/tests/integration/targets/sns_topic/aliases
+++ b/tests/integration/targets/sns_topic/aliases
@@ -1,4 +1 @@
-# reason: missing-policy
-unsupported
-
 cloud/aws
diff --git a/tests/integration/targets/sns_topic/defaults/main.yml b/tests/integration/targets/sns_topic/defaults/main.yml
index 19d84a3c5ea..c0134d0a286 100644
--- a/tests/integration/targets/sns_topic/defaults/main.yml
+++ b/tests/integration/targets/sns_topic/defaults/main.yml
@@ -1,12 +1,14 @@
-sns_topic_topic_name: "{{ resource_prefix }}-topic"
+# we hash the resource_prefix to get a shorter, unique string
+unique_id: "{{ resource_prefix | hash('md5') }}"
+
+sns_topic_topic_name: "ansible-test-{{ unique_id }}-topic"
 sns_topic_subscriptions:
   - endpoint: "{{ sns_topic_subscriber_arn }}"
     protocol: "lambda"
 sns_topic_third_party_topic_arn: "arn:aws:sns:us-east-1:806199016981:AmazonIpSpaceChanged"
 sns_topic_third_party_region: "{{ sns_topic_third_party_topic_arn.split(':')[3] }}"
+
+# additional test resource namings
 sns_topic_lambda_function: "sns_topic_lambda"
-sns_topic_lambda_name: "{{ resource_prefix }}-{{ sns_topic_lambda_function }}"
-# IAM role names have to be less than 64 characters
-# we hash the resource_prefix to get a shorter, unique string
-unique_id: "{{ resource_prefix | hash('md5') }}"
+sns_topic_lambda_name: "ansible-test-{{ unique_id }}-{{ sns_topic_lambda_function }}"
 sns_topic_lambda_role: "ansible-test-{{ unique_id }}-sns-lambda"
diff --git a/tests/integration/targets/sns_topic/tasks/main.yml b/tests/integration/targets/sns_topic/tasks/main.yml
index 4d494f2e8c7..891b162e4d2 100644
--- a/tests/integration/targets/sns_topic/tasks/main.yml
+++ b/tests/integration/targets/sns_topic/tasks/main.yml
@@ -7,7 +7,8 @@
   collections:
     - community.general
   block:
-  - name: create minimal lambda role
+
+  - name: create minimal lambda role (needed for subscription test further down)
     iam_role:
       name: '{{ sns_topic_lambda_role }}'
       assume_role_policy_document: '{{ lookup("file", "lambda-trust-policy.json") }}'
@@ -15,33 +16,65 @@
       managed_policies:
       - 'arn:aws:iam::aws:policy/AWSXrayWriteOnlyAccess'
     register: iam_role
+
   - name: pause if role was created
     pause:
       seconds: 10
     when: iam_role is changed
 
-  - name: create topic
+  - name: create standard SNS topic
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My topic name
     register: sns_topic_create
+
   - name: assert that creation worked
     assert:
       that:
       - sns_topic_create.changed
+
   - name: set sns_arn fact
     set_fact:
       sns_arn: '{{ sns_topic_create.sns_arn }}'
+
   - name: create topic again (expect changed=False)
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My topic name
     register: sns_topic_no_change
+
   - name: assert that recreation had no effect
     assert:
       that:
       - not sns_topic_no_change.changed
       - sns_topic_no_change.sns_arn == sns_topic_create.sns_arn
+
+  - name: Create a FIFO topic (not providing .fifo suffix, should be done automatically)
+    sns_topic:
+      name: '{{ sns_topic_topic_name }}-fifo'
+      topic_type: fifo
+      display_name: My FIFO topic
+    register: sns_fifo_topic
+
+  - name: assert that FIFO SNS topic creation worked
+    assert:
+      that:
+      - sns_fifo_topic.changed
+      - sns_fifo_topic.sns_topic.topic_type == 'fifo'
+      - sns_fifo_topic.sns_topic.name == '{{ sns_topic_topic_name }}-fifo.fifo'
+
+  - name: Run create a FIFO topic again for idempotence test
+    sns_topic:
+      name: '{{ sns_topic_topic_name }}-fifo.fifo'
+      topic_type: fifo
+      display_name: My FIFO topic
+    register: sns_fifo_topic
+
+  - name: assert that FIFO SNS topic creation worked
+    assert:
+      that:
+      - not sns_fifo_topic.changed
+
   - name: update display name
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -52,7 +85,8 @@
       that:
       - sns_topic_update_name.changed
       - sns_topic_update_name.sns_topic.display_name == "My new topic name"
-  - name: add policy
+
+  - name: add access policy to SNS topic
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My new topic name
@@ -62,17 +96,20 @@
     assert:
       that:
       - sns_topic_add_policy.changed
+
   - name: rerun same policy
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My new topic name
       policy: '{{ lookup(''template'', ''initial-policy.json'') }}'
     register: sns_topic_rerun_policy
+
   - name: assert that rerunning policy had no effect
     assert:
       that:
       - not sns_topic_rerun_policy.changed
-  - name: update policy
+
+  - name: update SNS policy
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My new topic name
@@ -82,6 +119,7 @@
     assert:
       that:
       - sns_topic_update_policy.changed
+
   - name: add delivery policy
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -97,6 +135,7 @@
             numMinDelayRetries: 0
             backoffFunction: linear
     register: sns_topic_add_delivery_policy
+
   - name: assert that adding delivery policy worked
     vars:
       delivery_policy: '{{ sns_topic_add_delivery_policy.sns_topic.delivery_policy | from_json }}'
@@ -106,6 +145,7 @@
       - delivery_policy.http.defaultHealthyRetryPolicy.minDelayTarget == 20
       - delivery_policy.http.defaultHealthyRetryPolicy.maxDelayTarget == 20
       - delivery_policy.http.defaultHealthyRetryPolicy.numRetries == 3
+
   - name: rerun same delivery policy
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -121,6 +161,7 @@
             numMinDelayRetries: 0
             backoffFunction: linear
     register: sns_topic_rerun_delivery_policy
+
   - name: assert that rerunning delivery_policy had no effect
     vars:
       delivery_policy: '{{ sns_topic_rerun_delivery_policy.sns_topic.delivery_policy | from_json }}'
@@ -130,6 +171,7 @@
       - delivery_policy.http.defaultHealthyRetryPolicy.minDelayTarget == 20
       - delivery_policy.http.defaultHealthyRetryPolicy.maxDelayTarget == 20
       - delivery_policy.http.defaultHealthyRetryPolicy.numRetries == 3
+
   - name: rerun a slightly different delivery policy
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -145,6 +187,7 @@
             numMinDelayRetries: 0
             backoffFunction: linear
     register: sns_topic_rerun_delivery_policy
+
   - name: assert that rerunning delivery_policy worked
     vars:
       delivery_policy: '{{ sns_topic_rerun_delivery_policy.sns_topic.delivery_policy | from_json }}'
@@ -154,15 +197,18 @@
       - delivery_policy.http.defaultHealthyRetryPolicy.minDelayTarget == 40
       - delivery_policy.http.defaultHealthyRetryPolicy.maxDelayTarget == 40
       - delivery_policy.http.defaultHealthyRetryPolicy.numRetries == 6
+
   - name: create temp dir
     tempfile:
       state: directory
     register: tempdir
+
   - name: ensure zip file exists
     archive:
       path: '{{ lookup(''first_found'', sns_topic_lambda_function) }}'
       dest: '{{ tempdir.path }}/{{ sns_topic_lambda_function }}.zip'
       format: zip
+
   - name: create lambda for subscribing (only auto-subscribing target available)
     lambda:
       name: '{{ sns_topic_lambda_name }}'
@@ -172,8 +218,10 @@
       role: '{{ sns_topic_lambda_role }}'
       handler: '{{ sns_topic_lambda_function }}.handler'
     register: lambda_result
+
   - set_fact:
       sns_topic_subscriber_arn: '{{ lambda_result.configuration.function_arn }}'
+
   - name: subscribe to topic
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -181,22 +229,26 @@
       purge_subscriptions: false
       subscriptions: '{{ sns_topic_subscriptions }}'
     register: sns_topic_subscribe
+
   - name: assert that subscribing worked
     assert:
       that:
       - sns_topic_subscribe.changed
       - sns_topic_subscribe.sns_topic.subscriptions|length == 1
+
   - name: run again with purge_subscriptions set to false
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       display_name: My new topic name
       purge_subscriptions: false
     register: sns_topic_no_purge
+
   - name: assert that not purging subscriptions had no effect
     assert:
       that:
       - not sns_topic_no_purge.changed
       - sns_topic_no_purge.sns_topic.subscriptions|length == 1
+
   - name: run again with purge_subscriptions set to true
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
@@ -208,26 +260,31 @@
       that:
       - sns_topic_purge.changed
       - sns_topic_purge.sns_topic.subscriptions|length == 0
+
   - name: delete topic
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       state: absent
+
   - name: no-op with third party topic (effectively get existing subscriptions)
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
       region: '{{ sns_topic_third_party_region }}'
     register: third_party_topic
+
   - name: subscribe to third party topic
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
       subscriptions: '{{ sns_topic_subscriptions }}'
       region: '{{ sns_topic_third_party_region }}'
     register: third_party_topic_subscribe
+
   - name: assert that subscribing worked
     assert:
       that:
       - third_party_topic_subscribe is changed
       - (third_party_topic_subscribe.sns_topic.subscriptions|length) - (third_party_topic.sns_topic.subscriptions|length) == 1
+
   - name: attempt to change name of third party topic
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
@@ -236,21 +293,25 @@
       region: '{{ sns_topic_third_party_region }}'
     ignore_errors: true
     register: third_party_name_change
+
   - name: assert that attempting to change display name does not work
     assert:
       that:
       - third_party_name_change is failed
+
   - name: unsubscribe from third party topic (purge_subscription defaults to true)
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
       subscriptions: '{{ third_party_topic.sns_topic.subscriptions }}'
       region: '{{ sns_topic_third_party_region }}'
     register: third_party_unsubscribe
+
   - name: assert that unsubscribing from third party topic works
     assert:
       that:
       - third_party_unsubscribe.changed
       - third_party_topic.sns_topic.subscriptions|length == third_party_unsubscribe.sns_topic.subscriptions|length
+
   - name: attempt to delete third party topic
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
@@ -259,25 +320,31 @@
       region: '{{ sns_topic_third_party_region }}'
     ignore_errors: true
     register: third_party_deletion
+
   - name: no-op after third party deletion
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
       region: '{{ sns_topic_third_party_region }}'
     register: third_party_deletion_facts
+
   - name: assert that attempting to delete third party topic does not work and preser
     assert:
       that:
       - third_party_deletion is failed
       - third_party_topic.sns_topic.subscriptions|length == third_party_deletion_facts.sns_topic.subscriptions|length
+
   always:
+
   - name: announce teardown start
     debug:
       msg: '************** TEARDOWN STARTS HERE *******************'
+
   - name: remove topic
     sns_topic:
       name: '{{ sns_topic_topic_name }}'
       state: absent
     ignore_errors: true
+
   - name: unsubscribe from third party topic
     sns_topic:
       name: '{{ sns_topic_third_party_topic_arn }}'
@@ -285,17 +352,20 @@
       purge_subscriptions: true
       region: '{{ sns_topic_third_party_region }}'
     ignore_errors: true
+
   - name: remove lambda
     lambda:
       name: '{{ sns_topic_lambda_name }}'
       state: absent
     ignore_errors: true
+
   - name: remove tempdir
     file:
       path: '{{ tempdir.path }}'
       state: absent
     when: tempdir is defined
     ignore_errors: true
+
   - name: remove iam role
     iam_role:
       name: '{{ sns_topic_lambda_role }}'