diff --git a/bulkwebhook/bulk_webhook/doctype/kafka_settings/confluent_kafka_utils.py b/bulkwebhook/bulk_webhook/doctype/kafka_settings/confluent_kafka_utils.py
index ebfb89e..9619837 100644
--- a/bulkwebhook/bulk_webhook/doctype/kafka_settings/confluent_kafka_utils.py
+++ b/bulkwebhook/bulk_webhook/doctype/kafka_settings/confluent_kafka_utils.py
@@ -35,7 +35,7 @@ def get_confluent_kafka_client(settings_doc):
     conf = {
         "bootstrap.servers": settings_doc.bootstrap_servers,
         "client.id": settings_doc.client_id,
-        "security.protocol": "SASL_SSL",
+        "security.protocol": settings_doc.security_protocol,
         "sasl.mechanism": "PLAIN",
         "sasl.username": settings_doc.get_password("api_key"),
         "sasl.password": settings_doc.get_password("api_secret"),
@@ -144,6 +144,10 @@ def run_kafka_hook_for_protobuf(kafka_hook, doctype, doc=None, doc_list=None):
         create_kafka_request_log(doctype=kafka_hook.webhook_doctype, status="sending to kafka", doc_list=doc_list)
 
     settings_doc = frappe.get_cached_doc("Kafka Settings", kafka_hook.kafka_settings)
+    if not settings_doc.security_protocol:
+        frappe.throw(
+            "Security Protocol is required for Kafka Settings, please set it on the Kafka Settings document."
+        )
     schema_registry_client = get_schema_registry_client(settings_doc)
     producer = get_confluent_kafka_producer(settings_doc)
 
diff --git a/bulkwebhook/bulk_webhook/doctype/kafka_settings/kafka_settings.json b/bulkwebhook/bulk_webhook/doctype/kafka_settings/kafka_settings.json
index 7c706cb..a1ca046 100755
--- a/bulkwebhook/bulk_webhook/doctype/kafka_settings/kafka_settings.json
+++ b/bulkwebhook/bulk_webhook/doctype/kafka_settings/kafka_settings.json
@@ -12,6 +12,8 @@
   "client_id",
   "api_key",
   "api_secret",
+  "column_break_9vzjb",
+  "security_protocol",
   "ksc_sec",
   "schema_regestry_url",
   "username",
@@ -69,11 +71,22 @@
    "fieldname": "password",
    "fieldtype": "Password",
    "label": "Password"
+  },
+  {
+   "fieldname": "column_break_9vzjb",
+   "fieldtype": "Column Break"
+  },
+  {
+   "fieldname": "security_protocol",
+   "fieldtype": "Select",
+   "label": "Security Protocol",
+   "options": "\nSSL\nSASL_SSL\nPLAINTEXT\nSASL_PLAINTEXT",
+   "reqd": 1
   }
  ],
  "index_web_pages_for_search": 1,
  "links": [],
- "modified": "2023-03-16 14:35:09.517429",
+ "modified": "2024-03-21 14:39:20.403911",
  "modified_by": "Administrator",
  "module": "Bulk Webhook",
  "name": "Kafka Settings",