diff --git a/salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup b/salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup
index aacc3ebc8..02624d813 100755
--- a/salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup
+++ b/salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup
@@ -79,18 +79,7 @@ printf "\n\n"
 
 printf "\nCreate Kafka Output Config if node is not an Import or Eval install\n"
 {% if grains.role not in ['so-import', 'so-eval'] %}
-KAFKACRT=$(openssl x509 -in /etc/pki/elasticfleet-kafka.crt)
-KAFKAKEY=$(openssl rsa -in  /etc/pki/elasticfleet-kafka.key)
-KAFKACA=$(openssl x509 -in  $INTCA)
-KAFKA_OUTPUT_VERSION="2.6.0"
-JSON_STRING=$( jq -n \
-                  --arg KAFKACRT "$KAFKACRT" \
-                  --arg KAFKAKEY "$KAFKAKEY" \
-                  --arg KAFKACA "$KAFKACA" \
-                  --arg KAFKA_OUTPUT_VERSION "$KAFKA_OUTPUT_VERSION" \
-                  '{ "name": "grid-kafka", "id": "so-manager_kafka", "type": "kafka", "hosts": [ "{{ GLOBALS.manager }}:9092", "{{ GLOBALS.manager_ip }}:9092" ], "is_default": false, "is_default_monitoring": false, "config_yaml": "", "ssl": { "certificate_authorities": [ $KAFKACA ], "certificate": $KAFKACRT, "key": $KAFKAKEY, "verification_mode": "full" }, "proxy_id": null, "client_id": "Elastic", "version": $KAFKA_OUTPUT_VERSION, "compression": "none", "auth_type": "ssl", "partition": "round_robin", "round_robin": { "group_events": 1 }, "topics":[{"topic":"%{[event.module]}-topic","when":{"type":"regexp","condition":"event.module:.+"}},{"topic":"default-topic"}], "headers": [ { "key": "", "value": "" } ], "timeout": 30, "broker_timeout": 30, "required_acks": 1 }'
-                  )
-curl -K /opt/so/conf/elasticsearch/curl.config -L -X POST "localhost:5601/api/fleet/outputs" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d "$JSON_STRING"
+salt-call state.apply kafka.elasticfleet queue=True
 {% endif %}
 
 # Add Manager Hostname & URL Base to Fleet Host URLs
diff --git a/salt/kafka/elasticfleet.sls b/salt/kafka/elasticfleet.sls
new file mode 100644
index 000000000..a91df765b
--- /dev/null
+++ b/salt/kafka/elasticfleet.sls
@@ -0,0 +1,22 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+
+{# Create Kafka output policy if it doesn't exist #}
+update_kafka_output_policy_script:
+  file.managed:
+    - name: /usr/sbin/so-kafka-fleet-output-policy
+    - source: salt://kafka/tools/sbin_jinja/so-kafka-fleet-output-policy
+    - user: root
+    - mode: 755
+    - template: jinja
+    - defaults:
+        GLOBALS: {{ GLOBALS }}
+
+create_kafka_output_policy:
+  cmd.run:
+    - name: 'so-kafka-fleet-output-policy > /dev/null 2>&1'
+    - show_changes: false
\ No newline at end of file
diff --git a/salt/kafka/init.sls b/salt/kafka/init.sls
index c4351ebfc..67b66c45d 100644
--- a/salt/kafka/init.sls
+++ b/salt/kafka/init.sls
@@ -10,6 +10,7 @@ include:
 {# Run kafka/nodes.sls before Kafka is enabled, so kafka nodes pillar is setup #}
 {% if grains.role in ['so-manager','so-managersearch', 'so-standalone'] %}
   - kafka.nodes
+  - kafka.elasticfleet
 {% endif %}
 {% if GLOBALS.pipeline == "KAFKA" and KAFKAMERGED.enabled %}
   - kafka.enabled
diff --git a/salt/kafka/tools/sbin_jinja/so-kafka-fleet-output-policy b/salt/kafka/tools/sbin_jinja/so-kafka-fleet-output-policy
new file mode 100644
index 000000000..13f158bdd
--- /dev/null
+++ b/salt/kafka/tools/sbin_jinja/so-kafka-fleet-output-policy
@@ -0,0 +1,23 @@
+#!/bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+output=$(curl -K /opt/so/conf/elasticsearch/curl.config -L "http://localhost:5601/api/fleet/outputs" | jq -r .items[].id)
+
+if ! echo "$output" | grep -q "so-manager_kafka"; then
+  KAFKACRT=$(openssl x509 -in /etc/pki/elasticfleet-kafka.crt)
+  KAFKAKEY=$(openssl rsa -in  /etc/pki/elasticfleet-kafka.key)
+  KAFKACA=$(openssl x509 -in  /etc/pki/tls/certs/intca.crt)
+  KAFKA_OUTPUT_VERSION="2.6.0"
+  JSON_STRING=$( jq -n \
+                --arg KAFKACRT "$KAFKACRT" \
+                --arg KAFKAKEY "$KAFKAKEY" \
+                --arg KAFKACA "$KAFKACA" \
+                --arg MANAGER_IP "{{ GLOBALS.manager_ip }}:9092" \
+                --arg KAFKA_OUTPUT_VERSION "$KAFKA_OUTPUT_VERSION" \
+                    '{ "name": "grid-kafka", "id": "so-manager_kafka", "type": "kafka", "hosts": [ $MANAGER_IP ], "is_default": false, "is_default_monitoring": false, "config_yaml": "", "ssl": { "certificate_authorities": [ $KAFKACA ], "certificate": $KAFKACRT, "key": $KAFKAKEY, "verification_mode": "full" }, "proxy_id": null, "client_id": "Elastic", "version": $KAFKA_OUTPUT_VERSION, "compression": "none", "auth_type": "ssl", "partition": "round_robin", "round_robin": { "group_events": 1 }, "topics":[{"topic":"%{[event.module]}-topic","when":{"type":"regexp","condition":"event.module:.+"}},{"topic":"default-topic"}], "headers": [ { "key": "", "value": "" } ], "timeout": 30, "broker_timeout": 30, "required_acks": 1 }'
+                )
+  curl -K /opt/so/conf/elasticsearch/curl.config -L -X POST "localhost:5601/api/fleet/outputs" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d "$JSON_STRING" 2&1> /dev/null
+fi
\ No newline at end of file