diff --git a/ansible/group_vars/all.yml b/ansible/group_vars/all.yml
index 99cf29c8c3..ba200a2464 100644
--- a/ansible/group_vars/all.yml
+++ b/ansible/group_vars/all.yml
@@ -1099,6 +1099,11 @@ infuxdb_internal_endpoint: "{{ internal_protocol }}://{{ kolla_internal_fqdn | p
 #################
 kafka_datadir_volume: "kafka"
 
+# The number of brokers in a Kafka cluster. This is used for automatically
+# setting quantities such as topic replicas and it is not recommended to
+# change it unless you know what you are doing.
+kafka_broker_count: "{{ groups['kafka'] | length }}"
+
 #########################
 # Internal Image options
 #########################
diff --git a/ansible/roles/kafka/defaults/main.yml b/ansible/roles/kafka/defaults/main.yml
index c3b4de5301..5a6a42e70a 100644
--- a/ansible/roles/kafka/defaults/main.yml
+++ b/ansible/roles/kafka/defaults/main.yml
@@ -20,7 +20,6 @@ kafka_cluster_name: "kolla_kafka"
 kafka_log_dir: "/var/log/kolla/kafka"
 kafka_heap_opts: "-Xmx1G -Xms1G"
 kafka_zookeeper: "{% for host in groups['zookeeper'] %}{{ 'api' | kolla_address(host) | put_address_in_context('url') }}:{{ zookeeper_client_port }}{% if not loop.last %},{% endif %}{% endfor %}"
-kafka_broker_count: "{{ groups['kafka'] | length }}"
 
 ####################
 # Docker
diff --git a/ansible/roles/monasca/defaults/main.yml b/ansible/roles/monasca/defaults/main.yml
index 24434e6670..5a4d506bb9 100644
--- a/ansible/roles/monasca/defaults/main.yml
+++ b/ansible/roles/monasca/defaults/main.yml
@@ -131,7 +131,22 @@ monasca_storm_nimbus_servers: "{% for host in groups['storm-nimbus'] %}'{{ 'api'
 # NOTE(dszumski): Only one NTP server is currently supported by the Monasca Agent plugin
 monasca_ntp_server: "{{ external_ntp_servers | first }}"
 
-# Kafka topics used by Monasca services
+# The default number of Kafka topic partitions. This effectively limits
+# the maximum number of workers per topic, counted over all nodes in the
+# Monasca deployment. For example, if you have a 3 node Monasca
+# deployment, you will by default have 3 instances of Monasca Persister,
+# with each instance having 2 workers by default for the metrics topic.
+# In this case, each worker on the metrics topic will be assigned 5
+# partitions of the metrics topic. If you increase the worker or instance
+# count, you may need to increase the partition count to ensure that all
+# workers can get a share of the work.
+monasca_default_topic_partitions: 30
+
+# The default number of topic replicas. Generally you should not change
+# this.
+monasca_default_topic_replication_factor: "{{ kafka_broker_count if kafka_broker_count|int < 3 else 3 }}"
+
+# Kafka topic names used by Monasca services
 monasca_metrics_topic: "metrics"
 monasca_raw_logs_topic: "logs"
 monasca_transformed_logs_topic: "transformed-logs"
@@ -141,6 +156,47 @@ monasca_alarm_notifications_topic: "alarm-notifications"
 monasca_alarm_notifications_retry_topic: "retry-notifications"
 monasca_periodic_notifications_topic: "60-seconds-notifications"
 
+# Kafka topic configuration. Most users will not need to modify these
+# settings, however for deployments where resources are tightly
+# constrained, or very large deployments where there are many parallel
+# workers, it is worth considering changing them. Note that if you do
+# change these settings, then you will need to manually remove each
+# topic from the Kafka deployment for the change to take effect when
+# the Monasca service is reconfigured.
+monasca_all_topics:
+  - name: "{{ monasca_metrics_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_raw_logs_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_transformed_logs_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_events_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_state_transitions_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_notifications_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_notifications_retry_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_periodic_notifications_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+
 # NOTE(dszumski): Due to the way monasca-notification is currently
 # implemented it is not recommended to change this period.
 monasca_periodic_notifications_period: 60
diff --git a/ansible/roles/monasca/tasks/bootstrap.yml b/ansible/roles/monasca/tasks/bootstrap.yml
index 2d0667b892..b116beb06d 100644
--- a/ansible/roles/monasca/tasks/bootstrap.yml
+++ b/ansible/roles/monasca/tasks/bootstrap.yml
@@ -61,6 +61,47 @@
   delegate_to: "{{ groups['influxdb'][0] }}"
   when: monasca_influxdb_name not in monasca_influxdb_database.stdout_lines
 
-# NOTE(dszumski): The Monasca APIs write logs and messages to Kafka. Since
-# Kafka has automatic topic generation enabled by default we don't need to
-# create topics here.
+# NOTE(dszumski): Although we can take advantage of automatic topic
+# creation in Kafka, creating the topics manually allows unique settings
+# to be used per topic, rather than the defaults. It also avoids an edge
+# case where services on multiple nodes may race to create topics, and
+# paves the way for enabling things like compression on a per topic basis.
+- name: List monasca kafka topics
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --zookeeper localhost
+    --list
+  register: kafka_topics
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+
+- name: Create monasca kafka topics if they don't exist
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --create
+    --topic {{ item.name }}
+    --partitions {{ item.partitions }}
+    --replication-factor {{ item.replication_factor }}
+    --zookeeper localhost
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+  when:
+    - item.name not in kafka_topics.stdout_lines
+    - item.enabled | bool
+  with_items: "{{ monasca_all_topics }}"
+
+- name: Remove monasca kafka topics for disabled services
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --delete
+    --topic "{{ item.name }}"
+    --zookeeper localhost
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+  when:
+    - item.name in kafka_topics.stdout_lines
+    - not item.enabled | bool
+  with_items: "{{ monasca_all_topics }}"
diff --git a/releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml b/releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml
new file mode 100644
index 0000000000..9b8bc81fe4
--- /dev/null
+++ b/releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml
@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    Adds support for explicitly creating individually customisable topics
+    in Kafka for Monasca.
+fixes:
+  - |
+    Fixes a trivial issue where some Monasca containers could momentarily
+    restart when initially racing to create topics in Kafka.