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..8ad782d 100644
--- a/apps/manual-kafka-cluster/group_vars/kafka/vars
+++ b/apps/manual-kafka-cluster/group_vars/kafka/vars
@@ -1,6 +1,6 @@
 # deployment vars
 ssh_keys:
-  - ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJQalZuAjeiWaPek5kJZxP4rTxuKlWgtSDFsdEGddf user1@desktop.local 
+  - ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIJQalZuAjeiWaPek5kJZxP4rTxuKlWgtSDFsdEGddf user1@desktop.local
   - ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAACAQCg2ANZcgWWFGh0vakgX1/d6Buea==
 
 instance_prefix: kafka
@@ -9,6 +9,12 @@ region: us-southeast
 image: linode/ubuntu24.04
 group:
 linode_tags:
+firewall_label:
+vpc_label:
+
+# Optional settings for DNS
+domain_name:
+ttl_sec:
 
 cluster_size: 3
 client_count: 2
@@ -26,4 +32,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..ccc9354 100644
--- a/apps/manual-kafka-cluster/provision.yml
+++ b/apps/manual-kafka-cluster/provision.yml
@@ -5,10 +5,116 @@
   vars_files:
       - group_vars/kafka/vars
       - group_vars/kafka/secret_vars
-  
+
   tasks:
 
+  - name: check if region supports all necessary features
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/region_info/
+    # https://techdocs.akamai.com/cloud-computing/docs/how-to-choose-a-data-center#product-availability
+    linode.cloud.region_info:
+      api_token: '{{ api_token }}'
+      id: '{{ region }}'
+    register: region_info
+
+  - name: assign vpc label
+    when: not vpc_label|d(False)
+    set_fact:
+      vpc_label: '{{ instance_prefix }}-cluster'
+
+  - name: assign vpc subnet
+    set_fact:
+      vpc_subnet: '10.0.0.0/{{ 32 - (((3*cluster_size) | log(2) | round(0, "ceil"))|int) }}'  # subnet supports an IP range 3x the cluster size
+
+  - name: set hosts per placement group
+    set_fact:
+      max_hosts_per_pg: 4
+    when: not max_hosts_per_pg|d(False)
+
+  - name: create vpc
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/vpc/
+    linode.cloud.vpc:
+      label: '{{ vpc_label }}-{{ region }}'
+      region: '{{ region }}'
+      api_token: '{{ api_token }}'
+      description: VPC for {{ linode_tags }} kafka brokers in {{ region }} to talk to each other.
+      state: present
+    register: vpc_info
+    when: vpc_label|d(False)
+
+  - name: get vpc subnet info
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/vpc_subnet_info/
+    linode.cloud.vpc_subnet_info:
+      label: '{{ vpc_label }}-{{ region }}-brokers'
+      vpc_id: '{{ vpc_info.vpc.id }}'
+      api_token: '{{ api_token }}'
+    register: vpc_subnet_info
+    ignore_errors: true
+    when:
+      - vpc_label|d(False)
+      - vpc_info.vpc.id
+
+  - name: remove incompatible subnet
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/vpc_subnet/
+    linode.cloud.vpc_subnet:
+      label: '{{ vpc_label }}-{{ region }}-brokers'
+      api_token: '{{ api_token }}'
+      state: absent
+      vpc_id: '{{ vpc_info.vpc.id }}'
+    when:
+      - vpc_label|d(False)
+      - vpc_info.vpc.id
+      - vpc_subnet_info is succeeded
+      - vpc_subnet_info.subnet.ipv4 != vpc_subnet
+
+  - name: create vpc subnet
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/vpc_subnet/
+    linode.cloud.vpc_subnet:
+      ipv4: '10.0.0.0/{{ 32 - (((3*cluster_size) | log(2) | round(0, "ceil"))|int) }}'  # subnet supports an IP range 3x the cluster size
+      label: '{{ vpc_label }}-{{ region }}-brokers'
+      api_token: '{{ api_token }}'
+      state: present
+      vpc_id: '{{ vpc_info.vpc.id }}'
+    register: vpc_subnet_info
+    when:
+      - vpc_label|d(False)
+      - vpc_info.vpc.id
+
+  - 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: create placement group for HA
+    # https://galaxy.ansible.com/ui/repo/published/linode/cloud/content/module/placement_group/
+    linode.cloud.placement_group:
+      label: '{{ instance_prefix }}-{{ region }}-ha-{{ item }}'
+      placement_group_policy: strict
+      placement_group_type: anti_affinity:local
+      region: '{{ region }}'
+      state: present
+      api_token: '{{ api_token }}'
+    register: placement_group_info
+    with_sequence: count='{{ (cluster_size / max_hosts_per_pg) | round(0, "ceil") | int }}'
+
+  - 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 }}'
@@ -17,11 +123,24 @@
       image: '{{ image }}'
       root_pass: '{{ root_password }}'
       authorized_keys: '{{ ssh_keys }}'
-      private_ip: true
-      ua_prefix: 'docs-kafka-occ'          
+      interfaces:
+        - purpose: 'vpc'
+          subnet_id: '{{ vpc_subnet_info.subnet.id }}'
+          primary: true
+          ipv4:
+            nat_1_1: 'any'
+      placement_group:
+        compliant_only: true
+        id: '{{ placement_group.id }}'
+      private_ip: false
+      ua_prefix: 'docs-kafka-occ'
       tags: '{{ linode_tags }}'
       state: present
+      firewall_id: '{{ (firewall_info.firewall|default({})).id|default(omit) }}'
+    vars:
+      placement_group: '{{ placement_group_info.results[(item|int-1) // max_hosts_per_pg].placement_group }}'
     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 +149,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,11 +197,11 @@
         kafka_data:
           server:
             {%- for count in range(cluster_size) %}
-            - kafka{{ count + 1 }}:
+            - name: {{ info.results[count].instance.label }}
               instance:
-                hostname: kafka{{ count + 1 }}
-                ip_pub1: {{ info.results[count].instance.ipv4[0] }}
-                ip_priv1: {{ info.results[count].instance.ipv4[1] }}
+                hostname: {{ info.results[count].instance.label }}{% if domain_name|d(False) and domain_info %}.{{ domain_name }}{% endif %}
+                ip_pub1: {{ info.results[count].networking.ipv4.public.0.address }}
+                ip_priv1: {{ info.results[count].networking.ipv4.vpc.0.address }}
             {%- endfor %}
 
   - name: add kafka nodes to inventory
@@ -54,13 +212,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