diff --git a/lib/cfnguardian/compile.rb b/lib/cfnguardian/compile.rb
index 0c349c4..c6c3a02 100644
--- a/lib/cfnguardian/compile.rb
+++ b/lib/cfnguardian/compile.rb
@@ -30,6 +30,8 @@
 require 'cfnguardian/resources/port'
 require 'cfnguardian/resources/internal_port'
 require 'cfnguardian/resources/nrpe'
+require 'cfnguardian/resources/kafka_cluster'
+require 'cfnguardian/resources/kafka_topic'
 require 'cfnguardian/resources/lambda'
 require 'cfnguardian/resources/network_targetgroup'
 require 'cfnguardian/resources/rds_cluster'
diff --git a/lib/cfnguardian/models/alarm.rb b/lib/cfnguardian/models/alarm.rb
index 0e22d4b..6c6a12f 100644
--- a/lib/cfnguardian/models/alarm.rb
+++ b/lib/cfnguardian/models/alarm.rb
@@ -394,6 +394,34 @@ def initialize(resource,environment)
       end
     end
 
+    class KafkaClusterAlarm < BaseAlarm
+      def initialize(resource,broker)
+        super(resource)
+        @group = 'KafkaCluster'
+        @namespace = 'AWS/Kafka'
+        @dimensions = { 'Cluster Name': resource['Id'], 'Broker ID': broker }
+        @statistic = 'Average'
+        @evaluation_periods = 1
+        @datapoints_to_alarm = 1
+        @period = 300
+        @treat_missing_data = 'breaching'
+      end
+    end
+
+    class KafkaTopicAlarm < BaseAlarm
+      def initialize(resource,broker)
+        super(resource)
+        @group = 'KafkaTopic'
+        @namespace = 'AWS/Kafka'
+        @dimensions = { 'Cluster Name': resource['ClusterName'], 'Broker ID': broker, Topic: resource['Id'] }
+        @statistic = 'Average'
+        @evaluation_periods = 1
+        @datapoints_to_alarm = 1
+        @period = 300
+        @treat_missing_data = 'breaching'
+      end
+    end
+
     class LambdaAlarm < BaseAlarm
       def initialize(resource)
         super(resource)
diff --git a/lib/cfnguardian/resources/kafka_cluster.rb b/lib/cfnguardian/resources/kafka_cluster.rb
new file mode 100644
index 0000000..eb88a53
--- /dev/null
+++ b/lib/cfnguardian/resources/kafka_cluster.rb
@@ -0,0 +1,74 @@
+module CfnGuardian::Resource
+  class KafkaCluster < Base
+
+    def initialize(resource, override_group = nil)
+      super(resource, override_group)
+      @brokers_list = resource['Brokers']
+    end
+
+    def default_alarms
+      @brokers_list.each do |broker|
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-CPUUserCritical"
+        alarm.metric_name = 'CpuUser'
+        alarm.threshold = 80
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-CPUUserWarning"
+        alarm.metric_name = 'CpuUser'
+        alarm.threshold = 50
+        alarm.alarm_action = 'Warning'
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-KafkaDataLogsDiskUsedCritical"
+        alarm.metric_name = 'KafkaDataLogsDiskUsed'
+        alarm.threshold = 85
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-KafkaDataLogsDiskUsedWarning"
+        alarm.metric_name = 'KafkaDataLogsDiskUsed'
+        alarm.threshold = 70
+        alarm.alarm_action = 'Warning'
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-BurstBalance"
+        alarm.metric_name = 'BurstBalance'
+        alarm.threshold = 1
+        alarm.comparison_operator = 'LessThanThreshold'
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-MemoryFreeCritical"
+        alarm.metric_name = 'MemoryFree'
+        alarm.threshold = 10
+        alarm.comparison_operator = 'LessThanThreshold'
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-MemoryFreeWarning"
+        alarm.metric_name = 'MemoryFree'
+        alarm.threshold = 50
+        alarm.alarm_action = 'Warning'
+        alarm.comparison_operator = 'LessThanThreshold'
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-NetworkRxErrorsCritical"
+        alarm.metric_name = 'NetworkRxErrors'
+        alarm.threshold = 10
+        @alarms.push(alarm)
+
+        alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-NetworkRxErrorsWarning"
+        alarm.metric_name = 'NetworkRxErrors'
+        alarm.threshold = 5
+        alarm.alarm_action = 'Warning'
+        @alarms.push(alarm)
+      end
+    end
+  end
+end
diff --git a/lib/cfnguardian/resources/kafka_topic.rb b/lib/cfnguardian/resources/kafka_topic.rb
new file mode 100644
index 0000000..ca5e6cc
--- /dev/null
+++ b/lib/cfnguardian/resources/kafka_topic.rb
@@ -0,0 +1,20 @@
+module CfnGuardian::Resource
+  class KafkaTopic < Base
+
+    def initialize(resource, override_group = nil)
+      super(resource, override_group)
+      @brokers_list = resource['Brokers']
+    end
+
+    def default_alarms
+      @brokers_list.each do |broker|  
+        alarm = CfnGuardian::Models::KafkaTopicAlarm.new(@resource,broker)
+        alarm.name = "Broker#{broker}-MessagesInPerSec"
+        alarm.metric_name = 'MessagesInPerSec'
+        alarm.threshold = 5
+        alarm.comparison_operator = 'LessThanThreshold'
+        @alarms.push(alarm)
+      end
+    end
+  end
+end