diff --git a/.rubocop-http---shopify-github-io-ruby-style-guide-rubocop-yml b/.rubocop-http---shopify-github-io-ruby-style-guide-rubocop-yml
index c68248c29..e77a04344 100644
--- a/.rubocop-http---shopify-github-io-ruby-style-guide-rubocop-yml
+++ b/.rubocop-http---shopify-github-io-ruby-style-guide-rubocop-yml
@@ -195,6 +195,7 @@ Style/FrozenStringLiteralComment:
   SupportedStyles:
     - always
     - never
+  SafeAutoCorrect: true
 
 Style/GlobalVars:
   AllowedVariables: []
diff --git a/lib/krane/deploy_task.rb b/lib/krane/deploy_task.rb
index b95cede1f..ba9557633 100644
--- a/lib/krane/deploy_task.rb
+++ b/lib/krane/deploy_task.rb
@@ -57,7 +57,7 @@ class DeployTask
     )
 
     def predeploy_sequence
-      default_group = { group: nil }
+      default_group = { groups: [], skip_groups: [] }
       before_crs = %w(
         ResourceQuota
         NetworkPolicy
@@ -71,10 +71,20 @@ def predeploy_sequence
 
       after_crs = %w(
         Pod
-      ).map { |r| [r, default_group] }
+      )
+
+      predeploy_hash = Hash[before_crs]
+      cluster_resource_discoverer.crds.select(&:predeployed?).each do |cr|
+        predeploy_hash[cr.kind] ||= { groups: [cr.group], skip_groups: [] }
+      end
+
+      after_crs.each { |cr| predeploy_hash[cr] = default_group }
+
+      cluster_resource_discoverer.crds.reject(&:predeployed?).each do |cr|
+        predeploy_hash[cr.kind][:skip_groups] << cr.group if predeploy_hash[cr.kind]
+      end
 
-      crs = cluster_resource_discoverer.crds.select(&:predeployed?).map { |cr| [cr.kind, { group: cr.group }] }
-      Hash[before_crs + crs + after_crs]
+      predeploy_hash
     end
 
     def prune_whitelist
@@ -216,7 +226,7 @@ def ejson_provisioners
     def deploy_has_priority_resources?(resources)
       resources.any? do |r|
         next unless (pr = predeploy_sequence[r.type])
-        !pr[:group] || pr[:group] == r.group
+        (pr[:groups].empty? || pr[:groups].include?(r.group)) && !pr[:skip_groups].include?(r.group)
       end
     end
 
diff --git a/lib/krane/kubernetes_resource.rb b/lib/krane/kubernetes_resource.rb
index a88c4f6a0..c202cd876 100644
--- a/lib/krane/kubernetes_resource.rb
+++ b/lib/krane/kubernetes_resource.rb
@@ -48,11 +48,11 @@ def build(namespace: nil, context:, definition:, logger:, statsd_tags:, crd: nil
         validate_definition_essentials(definition)
         opts = { namespace: namespace, context: context, definition: definition, logger: logger,
                  statsd_tags: statsd_tags }
-        if (klass = class_for_kind(definition["kind"]))
-          return klass.new(**opts)
-        end
-        if crd
+
+        if crd && definition['apiVersion'].start_with?("#{crd.group}/")
           CustomResource.new(crd: crd, **opts)
+        elsif (klass = class_for_kind(definition["kind"]))
+          klass.new(**opts)
         else
           type = definition["kind"]
           inst = new(**opts)
@@ -164,12 +164,16 @@ def file_path
     end
 
     def sync(cache)
-      @instance_data = cache.get_instance(kubectl_resource_type, name, raise_if_not_found: true)
+      @instance_data = cache.get_instance(sync_group_kind, name, raise_if_not_found: true)
     rescue Krane::Kubectl::ResourceNotFoundError
       @disappeared = true if deploy_started?
       @instance_data = {}
     end
 
+    def sync_group_kind
+      "#{kubectl_resource_type}.#{group}"
+    end
+
     def after_sync
     end
 
diff --git a/lib/krane/kubernetes_resource/custom_resource_definition.rb b/lib/krane/kubernetes_resource/custom_resource_definition.rb
index de0bf0c89..b2d103b79 100644
--- a/lib/krane/kubernetes_resource/custom_resource_definition.rb
+++ b/lib/krane/kubernetes_resource/custom_resource_definition.rb
@@ -87,6 +87,11 @@ def validate_rollout_conditions
       @rollout_conditions_validated = true
     end
 
+    def sync_group_kind
+      real_group = @definition.dig("apiVersion").split("/").first
+      "#{self.class.kind}.#{real_group}"
+    end
+
     private
 
     def names_accepted_condition
diff --git a/lib/krane/kubernetes_resource/daemon_set.rb b/lib/krane/kubernetes_resource/daemon_set.rb
index b296f633e..5cfe53665 100644
--- a/lib/krane/kubernetes_resource/daemon_set.rb
+++ b/lib/krane/kubernetes_resource/daemon_set.rb
@@ -3,7 +3,7 @@
 module Krane
   class DaemonSet < PodSetBase
     TIMEOUT = 5.minutes
-    SYNC_DEPENDENCIES = %w(Pod)
+    SYNC_DEPENDENCIES = %w(Pod.apps)
     attr_reader :pods
 
     def sync(cache)
diff --git a/lib/krane/kubernetes_resource/deployment.rb b/lib/krane/kubernetes_resource/deployment.rb
index b7ed43b9f..0dfcf50fb 100644
--- a/lib/krane/kubernetes_resource/deployment.rb
+++ b/lib/krane/kubernetes_resource/deployment.rb
@@ -4,7 +4,7 @@
 module Krane
   class Deployment < KubernetesResource
     TIMEOUT = 7.minutes
-    SYNC_DEPENDENCIES = %w(Pod ReplicaSet)
+    SYNC_DEPENDENCIES = %w(Pod.apps ReplicaSet.apps)
     REQUIRED_ROLLOUT_ANNOTATION = "required-rollout"
     REQUIRED_ROLLOUT_TYPES = %w(maxUnavailable full none).freeze
     DEFAULT_REQUIRED_ROLLOUT = 'full'
diff --git a/lib/krane/kubernetes_resource/replica_set.rb b/lib/krane/kubernetes_resource/replica_set.rb
index bb585bcef..204ac6dbe 100644
--- a/lib/krane/kubernetes_resource/replica_set.rb
+++ b/lib/krane/kubernetes_resource/replica_set.rb
@@ -4,7 +4,7 @@
 module Krane
   class ReplicaSet < PodSetBase
     TIMEOUT = 5.minutes
-    SYNC_DEPENDENCIES = %w(Pod)
+    SYNC_DEPENDENCIES = %w(Pod.apps)
     attr_reader :pods
 
     def initialize(namespace:, context:, definition:, logger:, statsd_tags: nil,
diff --git a/lib/krane/kubernetes_resource/service.rb b/lib/krane/kubernetes_resource/service.rb
index 78a7f35c3..dbe84f82a 100644
--- a/lib/krane/kubernetes_resource/service.rb
+++ b/lib/krane/kubernetes_resource/service.rb
@@ -4,7 +4,7 @@
 module Krane
   class Service < KubernetesResource
     TIMEOUT = 7.minutes
-    SYNC_DEPENDENCIES = %w(Pod Deployment StatefulSet)
+    SYNC_DEPENDENCIES = %w(Pod.apps Deployment.apps StatefulSet.apps)
 
     def sync(cache)
       super
diff --git a/lib/krane/kubernetes_resource/stateful_set.rb b/lib/krane/kubernetes_resource/stateful_set.rb
index e500a7271..ad8f2b3e6 100644
--- a/lib/krane/kubernetes_resource/stateful_set.rb
+++ b/lib/krane/kubernetes_resource/stateful_set.rb
@@ -4,7 +4,7 @@ module Krane
   class StatefulSet < PodSetBase
     TIMEOUT = 10.minutes
     ONDELETE = 'OnDelete'
-    SYNC_DEPENDENCIES = %w(Pod)
+    SYNC_DEPENDENCIES = %w(Pod.apps)
     attr_reader :pods
 
     def sync(cache)
diff --git a/lib/krane/resource_cache.rb b/lib/krane/resource_cache.rb
index baba9f8a8..d4e955c89 100644
--- a/lib/krane/resource_cache.rb
+++ b/lib/krane/resource_cache.rb
@@ -38,7 +38,7 @@ def get_all(kind, selector = nil)
 
     def prewarm(resources)
       sync_dependencies = resources.flat_map { |r| r.class.const_get(:SYNC_DEPENDENCIES) }
-      kinds = (resources.map(&:type) + sync_dependencies).uniq
+      kinds = (resources.map(&:sync_group_kind) + sync_dependencies).uniq
       Krane::Concurrency.split_across_threads(kinds, max_threads: kinds.count) { |kind| get_all(kind) }
     end
 
@@ -56,8 +56,12 @@ def use_or_populate_cache(kind)
     end
 
     def fetch_by_kind(kind)
-      resource_class = KubernetesResource.class_for_kind(kind)
-      global_kind = @task_config.global_kinds.map(&:downcase).include?(kind.downcase)
+      kind_only, group = kind.downcase.split(".", 2)
+      global_kind = @task_config.global_resources.any? do |resource|
+        resource['kind'].downcase == kind_only && resource['apigroup'].downcase == group
+      end
+
+      resource_class = KubernetesResource.class_for_kind(kind_only)
       output_is_sensitive = resource_class.nil? ? false : resource_class::SENSITIVE_TEMPLATE_CONTENT
       raw_json, _, st = @kubectl.run("get", kind, "--chunk-size=0", attempts: 5, output: "json",
          output_is_sensitive: output_is_sensitive, use_namespace: !global_kind)
diff --git a/lib/krane/resource_deployer.rb b/lib/krane/resource_deployer.rb
index 3990fb0af..0d9eed4c1 100644
--- a/lib/krane/resource_deployer.rb
+++ b/lib/krane/resource_deployer.rb
@@ -49,7 +49,8 @@ def predeploy_priority_resources(resource_list, predeploy_sequence)
       predeploy_sequence.each do |resource_type, attributes|
         matching_resources = resource_list.select do |r|
           r.type == resource_type &&
-          (!attributes[:group] || r.group == attributes[:group])
+          ((attributes[:groups].empty? || attributes[:groups].include?(r.group)) &&
+            !attributes[:skip_groups].include?(r.group))
         end
         StatsD.client.gauge('priority_resources.count', matching_resources.size, tags: statsd_tags)
 
diff --git a/lib/krane/task_config.rb b/lib/krane/task_config.rb
index 113304914..415bfa2d6 100644
--- a/lib/krane/task_config.rb
+++ b/lib/krane/task_config.rb
@@ -14,9 +14,13 @@ def initialize(context, namespace, logger = nil, kubeconfig = nil)
     end
 
     def global_kinds
-      @global_kinds ||= begin
+      global_resources.map { |g| g["kind"] }
+    end
+
+    def global_resources
+      @global_resources ||= begin
         cluster_resource_discoverer = ClusterResourceDiscovery.new(task_config: self)
-        cluster_resource_discoverer.fetch_resources(namespaced: false).map { |g| g["kind"] }
+        cluster_resource_discoverer.fetch_resources(namespaced: false)
       end
     end
 
diff --git a/test/fixtures/crd/network_policy.yml b/test/fixtures/crd/network_policy.yml
new file mode 100644
index 000000000..fa6751f4f
--- /dev/null
+++ b/test/fixtures/crd/network_policy.yml
@@ -0,0 +1,13 @@
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: networkpolicies.stable.example.io
+  labels:
+    app: krane
+spec:
+  group: stable.example.io
+  version: v1
+  names:
+    kind: NetworkPolicy
+    plural: networkpolicies
+  scope: Namespaced
diff --git a/test/fixtures/crd/network_policy_cr.yml b/test/fixtures/crd/network_policy_cr.yml
new file mode 100644
index 000000000..cbb8f4679
--- /dev/null
+++ b/test/fixtures/crd/network_policy_cr.yml
@@ -0,0 +1,20 @@
+---
+apiVersion: "stable.example.io/v1"
+kind: NetworkPolicy
+metadata:
+  name: cr-np
+---
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: built-in-policy
+spec:
+  ingress:
+  - ports:
+    - port: 80
+      protocol: TCP
+  podSelector:
+    matchLabels:
+      app: nginx-ingress-controller
+  policyTypes:
+  - Ingress
diff --git a/test/integration-serial/serial_deploy_test.rb b/test/integration-serial/serial_deploy_test.rb
index 41a26ea5d..6130858bb 100644
--- a/test/integration-serial/serial_deploy_test.rb
+++ b/test/integration-serial/serial_deploy_test.rb
@@ -236,6 +236,27 @@ def test_cr_merging
     assert_deploy_success(result)
   end
 
+  def test_custom_resources_predeployed_correctly_with_non_unique_kinds
+    results = deploy_global_fixtures("crd", subset: %w(network_policy.yml)) do |f|
+      # revert name muning since we're explicitly testing conflicting names
+      np = f["network_policy.yml"]["CustomResourceDefinition"].first
+      np["metadata"]["name"] = "networkpolicies.stable.example.io"
+      np["spec"]["names"] = { "kind" => "NetworkPolicy", "plural" => "networkpolicies" }
+    end
+    assert_deploy_success(results)
+    reset_logger
+
+    result = deploy_fixtures("crd", subset: %w(network_policy_cr.yml))
+    assert_deploy_success(result)
+    assert_logs_match_all([
+      /Phase 3: Predeploying priority resources/,
+      /Successfully deployed in \d.\ds: NetworkPolicy\/built-in-policy, NetworkPolicy\/cr-np/,
+      /Phase 4: Deploying all resources/,
+      /NetworkPolicy\/built-in-policy                     Created/,
+      /NetworkPolicy\/cr-np                               Exists/,
+    ], in_order: true)
+  end
+
   def test_custom_resources_predeployed
     assert_deploy_success(deploy_global_fixtures("crd", subset: %w(mail.yml things.yml widgets.yml)) do |f|
       mail = f.dig("mail.yml", "CustomResourceDefinition").first
diff --git a/test/unit/krane/resource_deployer_test.rb b/test/unit/krane/resource_deployer_test.rb
index 7c15e56c7..31a040918 100644
--- a/test/unit/krane/resource_deployer_test.rb
+++ b/test/unit/krane/resource_deployer_test.rb
@@ -71,7 +71,7 @@ def test_predeploy_priority_resources_respects_pre_deploy_list
     # are deployed. See test_predeploy_priority_resources_respects_empty_pre_deploy_list
     # for counter example
     Krane::ResourceWatcher.expects(:new).returns(watcher)
-    priority_list = { kind => { group: "core" } }
+    priority_list = { kind => { groups: ["core"], skip_groups: [] } }
     resource_deployer.predeploy_priority_resources([resource], priority_list)
   end