diff --git a/celery/canvas.py b/celery/canvas.py
index 866c1c888b2..f767de1ce0a 100644
--- a/celery/canvas.py
+++ b/celery/canvas.py
@@ -122,6 +122,9 @@ class Signature(dict):
 
     TYPES = {}
     _app = _type = None
+    # The following fields must not be changed during freezing/merging because
+    # to do so would disrupt completion of parent tasks
+    _IMMUTABLE_OPTIONS = {"group_id"}
 
     @classmethod
     def register_type(cls, name=None):
@@ -224,14 +227,22 @@ def apply_async(self, args=None, kwargs=None, route_name=None, **options):
     def _merge(self, args=None, kwargs=None, options=None, force=False):
         args = args if args else ()
         kwargs = kwargs if kwargs else {}
-        options = options if options else {}
+        if options is not None:
+            # We build a new options dictionary where values in `options`
+            # override values in `self.options` except for keys which are
+            # noted as being immutable (unrelated to signature immutability)
+            # implying that allowing their value to change would stall tasks
+            new_options = dict(self.options, **{
+                k: v for k, v in options.items()
+                if k not in self._IMMUTABLE_OPTIONS or k not in self.options
+            })
+        else:
+            new_options = self.options
         if self.immutable and not force:
-            return (self.args, self.kwargs,
-                    dict(self.options,
-                         **options) if options else self.options)
+            return (self.args, self.kwargs, new_options)
         return (tuple(args) + tuple(self.args) if args else self.args,
                 dict(self.kwargs, **kwargs) if kwargs else self.kwargs,
-                dict(self.options, **options) if options else self.options)
+                new_options)
 
     def clone(self, args=None, kwargs=None, **opts):
         """Create a copy of this signature.
@@ -286,7 +297,7 @@ def freeze(self, _id=None, group_id=None, chord=None,
             opts['parent_id'] = parent_id
         if 'reply_to' not in opts:
             opts['reply_to'] = self.app.oid
-        if group_id:
+        if group_id and "group_id" not in opts:
             opts['group_id'] = group_id
         if chord:
             opts['chord'] = chord
diff --git a/t/unit/tasks/test_canvas.py b/t/unit/tasks/test_canvas.py
index c15dec83d60..b90321572f3 100644
--- a/t/unit/tasks/test_canvas.py
+++ b/t/unit/tasks/test_canvas.py
@@ -1,5 +1,5 @@
 import json
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import MagicMock, Mock, patch, sentinel
 
 import pytest
 
@@ -154,6 +154,29 @@ def test_merge_immutable(self):
         assert kwargs == {'foo': 1}
         assert options == {'task_id': 3}
 
+    def test_merge_options__none(self):
+        sig = self.add.si()
+        _, _, new_options = sig._merge()
+        assert new_options is sig.options
+        _, _, new_options = sig._merge(options=None)
+        assert new_options is sig.options
+
+    @pytest.mark.parametrize("immutable_sig", (True, False))
+    def test_merge_options__group_id(self, immutable_sig):
+        # This is to avoid testing the behaviour in `test_set_immutable()`
+        if immutable_sig:
+            sig = self.add.si()
+        else:
+            sig = self.add.s()
+        # If the signature has no group ID, it can be set
+        assert not sig.options
+        _, _, new_options = sig._merge(options={"group_id": sentinel.gid})
+        assert new_options == {"group_id": sentinel.gid}
+        # But if one is already set, the new one is silently ignored
+        sig.set(group_id=sentinel.old_gid)
+        _, _, new_options = sig._merge(options={"group_id": sentinel.new_gid})
+        assert new_options == {"group_id": sentinel.old_gid}
+
     def test_set_immutable(self):
         x = self.add.s(2, 2)
         assert not x.immutable