diff --git a/lib/graphql/schema/mutation.rb b/lib/graphql/schema/mutation.rb
index 6d4a5f296f..b0ba8f3f5e 100644
--- a/lib/graphql/schema/mutation.rb
+++ b/lib/graphql/schema/mutation.rb
@@ -62,6 +62,13 @@ class Mutation < GraphQL::Schema::Resolver
       extend GraphQL::Schema::Member::HasFields
       extend GraphQL::Schema::Resolver::HasPayloadType
 
+      # @api private
+      def call_resolve(_args_hash)
+        # Clear any cached values from `loads` or authorization:
+        dataloader.clear_cache
+        super
+      end
+
       class << self
         def visible?(context)
           true
diff --git a/lib/graphql/schema/resolver.rb b/lib/graphql/schema/resolver.rb
index 7751af0c85..19da62a587 100644
--- a/lib/graphql/schema/resolver.rb
+++ b/lib/graphql/schema/resolver.rb
@@ -103,11 +103,7 @@ def resolve_with_support(**args)
                   end
                 elsif authorized_val
                   # Finally, all the hooks have passed, so resolve it
-                  if loaded_args.any?
-                    public_send(self.class.resolve_method, **loaded_args)
-                  else
-                    public_send(self.class.resolve_method)
-                  end
+                  call_resolve(loaded_args)
                 else
                   raise GraphQL::UnauthorizedFieldError.new(context: context, object: object, type: field.owner, field: field)
                 end
@@ -117,6 +113,15 @@ def resolve_with_support(**args)
         end
       end
 
+      # @api private {GraphQL::Schema::Mutation} uses this to clear the dataloader cache
+      def call_resolve(args_hash)
+        if args_hash.any?
+          public_send(self.class.resolve_method, **args_hash)
+        else
+          public_send(self.class.resolve_method)
+        end
+      end
+
       # Do the work. Everything happens here.
       # @return [Object] An object corresponding to the return type
       def resolve(**args)
diff --git a/spec/graphql/schema/mutation_spec.rb b/spec/graphql/schema/mutation_spec.rb
index 4815d7b9ba..6cd792120a 100644
--- a/spec/graphql/schema/mutation_spec.rb
+++ b/spec/graphql/schema/mutation_spec.rb
@@ -258,4 +258,75 @@ def resolve(**inputs)
     res = schema.execute("mutation { child(thingName: \"abc\", thingId: \"123\") { inputs } }")
     assert_equal "{:thing_id=>\"123\", :thing_name=>\"abc\"}", res["data"]["child"]["inputs"]
   end
+
+  describe "flushing dataloader cache" do
+    class MutationDataloaderCacheSchema < GraphQL::Schema
+      module Database
+        DATA = {}
+        def self.get(id)
+          value = DATA[id] ||= 0
+          OpenStruct.new(id: id, value: value)
+        end
+
+        def self.increment(id)
+          DATA[id] ||= 0
+          DATA[id] += 1
+        end
+
+        def self.clear
+          DATA.clear
+        end
+      end
+
+      class CounterSource < GraphQL::Dataloader::Source
+        def fetch(ids)
+          ids.map { |id| Database.get(id) }
+        end
+      end
+      class CounterType < GraphQL::Schema::Object
+        def self.authorized?(obj, ctx)
+          # Just force the load here, too:
+          ctx.dataloader.with(CounterSource).load(obj.id)
+          true
+        end
+        field :value, Integer
+      end
+      class Increment < GraphQL::Schema::Mutation
+        field :counter, CounterType
+        argument :counter_id, ID, loads: CounterType
+
+        def resolve(counter:)
+          Database.increment(counter.id)
+          {
+            counter: dataloader.with(CounterSource).load(counter.id)
+          }
+        end
+      end
+
+      class Mutation < GraphQL::Schema::Object
+        field :increment, mutation: Increment
+      end
+
+      mutation(Mutation)
+
+      def self.object_from_id(id, ctx)
+        ctx.dataloader.with(CounterSource).load(id)
+      end
+
+      def self.resolve_type(abs_type, obj, ctx)
+        CounterType
+      end
+
+      use GraphQL::Dataloader
+    end
+
+    it "clears the cache after authorized and loads" do
+      MutationDataloaderCacheSchema::Database.clear
+      res = MutationDataloaderCacheSchema.execute("mutation { increment(counterId: \"4\") { counter { value } } }")
+      assert_equal 1, res["data"]["increment"]["counter"]["value"]
+
+      res2 = MutationDataloaderCacheSchema.execute("mutation { increment(counterId: \"4\") { counter { value } } }")
+      assert_equal 2, res2["data"]["increment"]["counter"]["value"]
+    end
+  end
 end