diff --git a/salt/kafka/defaults.yaml b/salt/kafka/defaults.yaml
index c20d8552c..21d6956ba 100644
--- a/salt/kafka/defaults.yaml
+++ b/salt/kafka/defaults.yaml
@@ -3,6 +3,7 @@ kafka:
   cluster_id:
   controllers:
   reset:
+  logstash: []
   config:
     password:
     trustpass:
diff --git a/salt/kafka/soc_kafka.yaml b/salt/kafka/soc_kafka.yaml
index 872bf51f2..8087f9bdf 100644
--- a/salt/kafka/soc_kafka.yaml
+++ b/salt/kafka/soc_kafka.yaml
@@ -16,6 +16,12 @@ kafka:
     description: Disable and reset the Kafka cluster. This will remove all Kafka data including logs that may have not yet been ingested into Elasticsearch and reverts the grid to using REDIS as the global pipeline. This is useful when testing different Kafka configurations such as rearranging Kafka brokers / controllers allowing you to reset the cluster rather than manually fixing any issues arising from attempting to reassign a Kafka broker into a controller. Enter 'YES_RESET_KAFKA' and submit to disable and reset Kafka. Make any configuration changes required and re-enable Kafka when ready. This action CANNOT be reversed.
     advanced: True
     helpLink: kafka.html
+  logstash:
+    description: By default logstash is disabled when Kafka is enabled. This option allows you to specify any hosts you would like to re-enable logstash on alongside Kafka.
+    forcedType: "[]string"
+    multiline: True
+    advanced: True
+    helpLink: kafka.html
   config:
     password:
       description: The password used for the Kafka certificates.
diff --git a/salt/logstash/defaults.yaml b/salt/logstash/defaults.yaml
index b4bc885f6..9930b7bcf 100644
--- a/salt/logstash/defaults.yaml
+++ b/salt/logstash/defaults.yaml
@@ -25,7 +25,7 @@ logstash:
       - so/0011_input_endgame.conf
       - so/0012_input_elastic_agent.conf.jinja
       - so/0013_input_lumberjack_fleet.conf
-      - so/9999_output_redis.conf.jinja        
+      - so/9999_output_redis.conf.jinja
     receiver:
       - so/0011_input_endgame.conf
       - so/0012_input_elastic_agent.conf.jinja
@@ -35,7 +35,6 @@ logstash:
       - so/0900_input_redis.conf.jinja
       - so/9805_output_elastic_agent.conf.jinja
       - so/9900_output_endgame.conf.jinja
-      - so/0800_input_kafka.conf.jinja
     custom0: []
     custom1: []
     custom2: []
diff --git a/salt/logstash/init.sls b/salt/logstash/init.sls
index 3bc539b35..62b2a2ebb 100644
--- a/salt/logstash/init.sls
+++ b/salt/logstash/init.sls
@@ -4,13 +4,9 @@
 # Elastic License 2.0.
 
 {% from 'logstash/map.jinja' import LOGSTASH_MERGED %}
-{% from 'kafka/map.jinja' import KAFKAMERGED %}
 
 include:
-{# Disable logstash when Kafka is enabled except when the role is standalone #}
-{% if LOGSTASH_MERGED.enabled and grains.role == 'so-standalone' %}
-  - logstash.enabled
-{% elif LOGSTASH_MERGED.enabled and not KAFKAMERGED.enabled %}
+{% if LOGSTASH_MERGED.enabled %}
   - logstash.enabled
 {% else %}
   - logstash.disabled
diff --git a/salt/logstash/map.jinja b/salt/logstash/map.jinja
index 0bb6de2b7..2c3e02864 100644
--- a/salt/logstash/map.jinja
+++ b/salt/logstash/map.jinja
@@ -6,6 +6,7 @@
 {% from 'vars/globals.map.jinja' import GLOBALS %}
 {% import_yaml 'logstash/defaults.yaml' as LOGSTASH_DEFAULTS %}
 {% set LOGSTASH_MERGED = salt['pillar.get']('logstash', LOGSTASH_DEFAULTS.logstash, merge=True) %}
+{% set KAFKA_LOGSTASH = salt['pillar.get']('kafka:logstash', []) %}
 
 {# used to store the redis nodes that logstash needs to know about to pull from the queue #}
 {% set LOGSTASH_REDIS_NODES = [] %}
@@ -30,3 +31,14 @@
 {%   endfor %}
 {% endfor %}
 
+{# Append Kafka input pipeline when Kafka is enabled #}
+{% if GLOBALS.pipeline == 'KAFKA' %}
+{%   do LOGSTASH_MERGED.defined_pipelines.search.append('so/0800_input_kafka.conf.jinja') %}
+{%   do LOGSTASH_MERGED.defined_pipelines.manager.append('so/0800_input_kafka.conf.jinja') %}
+{# Disable logstash on manager & receiver nodes unless it has an override configured #}
+{%   if not KAFKA_LOGSTASH %}
+{%     if GLOBALS.role in ['so-manager', 'so-receiver'] and GLOBALS.hostname not in KAFKA_LOGSTASH %}
+{%       do LOGSTASH_MERGED.update({'enabled': False}) %}
+{%     endif %}
+{%   endif %}
+{% endif %}
\ No newline at end of file