diff --git a/apps/manual-kafka-cluster/collections.yml b/apps/manual-kafka-cluster/collections.yml
index 51bd91c..a4ed148 100644
--- a/apps/manual-kafka-cluster/collections.yml
+++ b/apps/manual-kafka-cluster/collections.yml
@@ -2,6 +2,6 @@ collections:
  - name: community.crypto
    version: 2.15.1
  - name: linode.cloud
-   version: 0.16.1 
+   version: 0.37.1
  - name: community.general
-   version: 8.6.0
\ No newline at end of file
+   version: 8.6.0
diff --git a/apps/manual-kafka-cluster/group_vars/kafka/vars b/apps/manual-kafka-cluster/group_vars/kafka/vars
index 0a8513b..c766d0a 100644
--- a/apps/manual-kafka-cluster/group_vars/kafka/vars
+++ b/apps/manual-kafka-cluster/group_vars/kafka/vars
@@ -9,6 +9,11 @@ region: us-southeast
 image: linode/ubuntu24.04
 group:
 linode_tags:
+firewall_label:
+
+# Optional settings for DNS
+domain_name:
+ttl_sec:
 
 cluster_size: 3
 client_count: 2
@@ -26,4 +31,4 @@ state_or_province_name: Pennsylvania
 locality_name: Philadelphia
 organization_name: Akamai Technologies
 email_address: webmaster@example.com
-ca_common_name: Kafka RootCA
\ No newline at end of file
+ca_common_name: Kafka RootCA
diff --git a/apps/manual-kafka-cluster/provision.yml b/apps/manual-kafka-cluster/provision.yml
index 05bfe76..f868c14 100644
--- a/apps/manual-kafka-cluster/provision.yml
+++ b/apps/manual-kafka-cluster/provision.yml
@@ -8,7 +8,30 @@
   
   tasks:
 
+  - name: get firewall info
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/firewall_info/
+    linode.cloud.firewall_info:
+      label: '{{ firewall_label }}'
+      api_token: '{{ api_token }}'
+    register: firewall_info
+    when: firewall_label|d(False)
+
+  - name: check if instances already created
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/instance_list/
+    linode.cloud.instance_list:
+      api_token: '{{ api_token }}'
+      filters:
+        - name: label
+          values: "{{ [ instance_prefix ] | product(range(1, cluster_size+1)) | map('join') | list }}"
+      order_by: label
+    register: existing_instances
+
+  - name: convert instances to a dict
+    set_fact:
+      existing_instances: "{{ dict(existing_instances.instances | map(attribute='label') | zip(existing_instances.instances)) }}"
+
   - name: creating kafka servers
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/instance/
     linode.cloud.instance:
       label: '{{ instance_prefix }}{{ item }}'
       api_token: '{{ api_token }}'
@@ -21,7 +44,9 @@
       ua_prefix: 'docs-kafka-occ'          
       tags: '{{ linode_tags }}'
       state: present
+      firewall_id: '{{ (firewall_info.firewall|default({})).id|default(omit) }}'
     with_sequence: count='{{ cluster_size }}'
+    when: (instance_prefix + item) not in existing_instances
 
   - name: get info about the instances
     linode.cloud.instance_info:
@@ -30,6 +55,45 @@
     register: info
     with_sequence: count='{{ cluster_size }}'
 
+  - name: check if hosts are in DNS
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/domain_info/
+    linode.cloud.domain_info:
+      api_token: '{{ api_token }}'
+      domain: '{{ domain_name }}'
+    when: domain_name|d(False)
+    register: domain_info
+
+  - name: remove old hosts from dns
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/domain_record/
+    linode.cloud.domain_record:
+      api_token: '{{ api_token }}'
+      domain_id: '{{ domain_info.domain.id }}'
+      record_id: '{{ item.id }}'
+      state: absent
+    vars:
+      instance_ips: "{{ dict(info.results | map(attribute='instance.label') | zip(info.results | map(attribute='networking.ipv4.public.0.address'))) }}"
+    when:
+      - domain_name|d(False)
+      - domain_info|d(False)
+      - item.name in instance_ips
+      - item.target != instance_ips[item.name]
+    with_items: "{{ domain_info.records }}"
+
+  - name: add new hosts to dns
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/domain_record/
+    linode.cloud.domain_record:
+      api_token: '{{ api_token }}'
+      domain_id: '{{ domain_info.domain.id }}'
+      name: '{{ item.instance.label }}'
+      target: '{{ item.instance.ipv4[0] }}'
+      ttl_sec: '{{ ttl_sec | default(omit) }}'
+      type: 'A'
+      state: present
+    with_items: "{{ info.results }}"
+    when:
+      - domain_name|d(False)
+      - domain_info|d(False)
+
   - name: update group_vars
     blockinfile:
       path: ./group_vars/kafka/vars
@@ -39,9 +103,9 @@
         kafka_data:
           server:
             {%- for count in range(cluster_size) %}
-            - kafka{{ count + 1 }}:
+            - name: {{ info.results[count].instance.label }}
               instance:
-                hostname: kafka{{ count + 1 }}
+                hostname: {{ info.results[count].instance.label }}{% if domain_name|d(False) and domain_info %}.{{ domain_name }}{% endif %}
                 ip_pub1: {{ info.results[count].instance.ipv4[0] }}
                 ip_priv1: {{ info.results[count].instance.ipv4[1] }}
             {%- endfor %}
@@ -54,13 +118,13 @@
         #jinja2: trim_blocks:False
         [kafka]
         {%- for count in range(cluster_size) %}
-        {{ info.results[count].instance.ipv4[0] }} {% if count < controller_count %}role='controller and broker'{%else%}role='broker only'{%endif%}
+        {{ info.results[count].networking.ipv4.public[0].rdns }} {% if count < controller_count %}role='controller and broker'{%else%}role='broker only'{%endif%}
         {%- endfor %}
 
   - name: wait for port 22 to become open
     wait_for:
       port: 22
-      host: '{{ item.instance.ipv4[0] }}'
+      host: '{{ item.networking.ipv4.public[0].rdns }}'
       search_regex: OpenSSH
       delay: 10
     connection: local
diff --git a/apps/manual-kafka-cluster/requirements.txt b/apps/manual-kafka-cluster/requirements.txt
index f2e0448..d8932ec 100644
--- a/apps/manual-kafka-cluster/requirements.txt
+++ b/apps/manual-kafka-cluster/requirements.txt
@@ -8,6 +8,6 @@ pyyaml==6.0.1
 dnspython==2.2.1
 passlib==1.7.4
 ## cloud.linode module dependancies ##
-linode-api4==5.15.1
+linode-api4==5.29.0
 polling==0.3.2
 ansible-specdoc==0.0.14
diff --git a/apps/manual-kafka-cluster/roles/common/tasks/main.yml b/apps/manual-kafka-cluster/roles/common/tasks/main.yml
index 1648f23..49104c0 100644
--- a/apps/manual-kafka-cluster/roles/common/tasks/main.yml
+++ b/apps/manual-kafka-cluster/roles/common/tasks/main.yml
@@ -46,3 +46,6 @@
 
 - name: apply ufw rules
   import_tasks: ufw_rules.yml
+
+- name: reboot linode
+  reboot:
diff --git a/apps/manual-kafka-cluster/roles/kafka/tasks/configure.yml b/apps/manual-kafka-cluster/roles/kafka/tasks/configure.yml
index 53124e4..3b1609b 100644
--- a/apps/manual-kafka-cluster/roles/kafka/tasks/configure.yml
+++ b/apps/manual-kafka-cluster/roles/kafka/tasks/configure.yml
@@ -51,16 +51,31 @@
     index_var: count
   when: hostvars[groups['kafka'][count]].role == 'broker only'
 
+- name: check existing kafka cluster uuid
+  shell:
+    cmd: grep -s cluster.id {{ kafka_data_directory }}/data/kraft-combined-logs/meta.properties | cut -f 2 -d = || /bin/true
+    removes: "{{ kafka_data_directory}}/data/kraft-combined-logs/meta.properties"
+  register: old_cluster_uuid
+  run_once: true
+  delegate_to: "{{ groups['kafka'][0] }}"
+
 - name: create kafka cluster uuid
   command:
     cmd:  "{{ kafka_bin_directory }}/kafka-storage.sh random-uuid"
-  register: cluster_uuid
+    creates: "{{ kafka_data_directory}}/data/kraft-combined-logs/meta.properties"
+  register: new_cluster_uuid
+  run_once: true
+  delegate_to: "{{ groups['kafka'][0] }}"
+
+- name: determine kafka cluster uuid
+  set_fact:
+    cluster_uuid: "{{ (old_cluster_uuid.changed) | ternary(old_cluster_uuid.stdout, new_cluster_uuid.stdout) }}"
   run_once: true
   delegate_to: "{{ groups['kafka'][0] }}"
 
 - name: format data directory for controller and broker nodes
   command:
-    cmd: "{{ kafka_bin_directory }}/kafka-storage.sh format -t {{ cluster_uuid.stdout }} -c {{ kafka_config_directory }}/config/kraft/server.properties"
+    cmd: "{{ kafka_bin_directory }}/kafka-storage.sh format -t {{ cluster_uuid}} -c {{ kafka_config_directory }}/config/kraft/server.properties --ignore-formatted"
   become: true
   become_user: kafka
   run_once: true
@@ -72,7 +87,7 @@
 
 - name: format data directory broker nodes
   command:
-    cmd: "{{ kafka_bin_directory }}/kafka-storage.sh format -t {{ cluster_uuid.stdout }} -c {{ kafka_config_directory }}/config/kraft/broker.properties"
+    cmd: "{{ kafka_bin_directory }}/kafka-storage.sh format -t {{ cluster_uuid}} -c {{ kafka_config_directory }}/config/kraft/broker.properties --ignore-formatted"
   become: true
   become_user: kafka
   run_once: true
diff --git a/apps/manual-kafka-cluster/roles/kafka/tasks/hostname.yml b/apps/manual-kafka-cluster/roles/kafka/tasks/hostname.yml
index 0f9d261..3086be8 100644
--- a/apps/manual-kafka-cluster/roles/kafka/tasks/hostname.yml
+++ b/apps/manual-kafka-cluster/roles/kafka/tasks/hostname.yml
@@ -8,7 +8,7 @@
     block: |
       #jinja2: trim_blocks:False
       {%- for count in range(cluster_size) %}
-      {{ kafka_data.server[count].instance.ip_priv1 }} {{ kafka_data.server[count].instance.hostname }}
+      {{ kafka_data.server[count].instance.ip_priv1 }} {{ kafka_data.server[count].instance.hostname }} {{ kafka_data.server[count].name }}
       {%- endfor %}
 
 - name: configure hostnames
@@ -19,4 +19,4 @@
   delegate_to: "{{ item }}"
   loop: "{{ groups['kafka'] }}"
   loop_control:
-    index_var: count
\ No newline at end of file
+    index_var: count
diff --git a/apps/manual-kafka-cluster/shutdown.yml b/apps/manual-kafka-cluster/shutdown.yml
new file mode 100644
index 0000000..0daf8e9
--- /dev/null
+++ b/apps/manual-kafka-cluster/shutdown.yml
@@ -0,0 +1,66 @@
+---
+# shutdown kafka server and client instances
+- name: shutdown kafka instances
+  hosts: localhost
+  vars_files:
+      - group_vars/kafka/vars
+      - group_vars/kafka/secret_vars
+
+  tasks:
+
+  # DNS
+
+  - name: check if hosts are in DNS
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/domain_info/
+    linode.cloud.domain_info:
+      api_token: '{{ api_token }}'
+      domain: '{{ domain_name }}'
+    when: domain_name|d(False)
+    register: domain_info
+
+  - name: remove hosts from dns
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/domain_record/
+    linode.cloud.domain_record:
+      api_token: '{{ api_token }}'
+      domain_id: '{{ domain_info.domain.id }}'
+      record_id: '{{ item.id }}'
+      state: absent
+    when:
+      - domain_name|d(False)
+      - domain_info|d(False)
+      - item.target in (kafka_data.server | map(attribute='instance.ip_pub1'))
+    with_items: "{{ domain_info.records }}"
+
+
+  # Instances
+
+  - name: get list of instances
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/instance_list/
+    linode.cloud.instance_list:
+      api_token: '{{ api_token }}'
+      filters:
+        - name: label
+          values: "{{ [ instance_prefix ] | product(range(1, cluster_size+1)) | map('join') | list }}"
+      order_by: label
+    register: existing_instances
+
+  - name: shutdown kafka servers
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/instance/
+    linode.cloud.instance:
+      label: '{{ item.label }}'
+      api_token: '{{ api_token }}'
+      region: '{{ region }}'
+      state: absent
+    with_items: '{{ existing_instances.instances }}'
+
+  - name: update group_vars
+    blockinfile:
+      path: ./group_vars/kafka/vars
+      marker: "# {mark} INSTANCE VARS"
+      state: absent
+
+  - name: remove kafka nodes from inventory
+    blockinfile:
+      path: ./hosts
+      marker: "# {mark} KAFKA INSTANCES"
+      state: absent