diff --git a/neutron/db/l3_db.py b/neutron/db/l3_db.py
index 008c6161971..40930ba6732 100644
--- a/neutron/db/l3_db.py
+++ b/neutron/db/l3_db.py
@@ -1907,6 +1907,22 @@ class L3RpcNotifierMixin(object):
         for router_id in router_ids:
             l3plugin.notify_router_updated(context, router_id)
 
+    @staticmethod
+    @registry.receives(resources.PORT, [events.AFTER_UPDATE])
+    def _notify_gateway_port_ip_changed(resource, event, trigger, **kwargs):
+        l3plugin = directory.get_plugin(plugin_constants.L3)
+        if not l3plugin:
+            return
+        new_port = kwargs.get('port')
+        original_port = kwargs.get('original_port')
+
+        if original_port['device_owner'] != constants.DEVICE_OWNER_ROUTER_GW:
+            return
+
+        if utils.port_ip_changed(new_port, original_port):
+            l3plugin.notify_router_updated(kwargs['context'],
+                                           new_port['device_id'])
+
     @staticmethod
     @registry.receives(resources.SUBNETPOOL_ADDRESS_SCOPE,
                        [events.AFTER_UPDATE])
diff --git a/neutron/tests/fullstack/resources/client.py b/neutron/tests/fullstack/resources/client.py
index 07176e59165..441c3e50661 100644
--- a/neutron/tests/fullstack/resources/client.py
+++ b/neutron/tests/fullstack/resources/client.py
@@ -118,6 +118,10 @@ class ClientFixture(fixtures.Fixture):
 
         return self._create_resource(resource_type, spec)
 
+    def list_ports(self, retrieve_all=True, **kwargs):
+        resp = self.client.list_ports(retrieve_all=retrieve_all, **kwargs)
+        return resp['ports']
+
     def create_port(self, tenant_id, network_id, hostname=None,
                     qos_policy_id=None, security_groups=None, **kwargs):
         spec = {
diff --git a/neutron/tests/fullstack/test_l3_agent.py b/neutron/tests/fullstack/test_l3_agent.py
index 04790652ef5..e0cbea07054 100644
--- a/neutron/tests/fullstack/test_l3_agent.py
+++ b/neutron/tests/fullstack/test_l3_agent.py
@@ -17,6 +17,7 @@ import os
 import time
 
 import netaddr
+from neutron_lib import constants
 from oslo_utils import uuidutils
 
 from neutron.agent.l3 import ha_router
@@ -76,6 +77,50 @@ class TestL3Agent(base.BaseFullStackTestCase):
 
         return self._boot_fake_vm_in_network(host, tenant_id, network['id'])
 
+    def _test_gateway_ip_changed(self):
+        tenant_id = uuidutils.generate_uuid()
+        ext_net, ext_sub = self._create_external_network_and_subnet(tenant_id)
+        external_vm = self.useFixture(
+            machine_fixtures.FakeMachine(
+                self.environment.central_bridge,
+                common_utils.ip_to_cidr(ext_sub['gateway_ip'], 24)))
+
+        router = self.safe_client.create_router(tenant_id,
+                                                external_network=ext_net['id'])
+
+        vm = self._create_net_subnet_and_vm(
+            tenant_id, ['20.0.0.0/24', '2001:db8:aaaa::/64'],
+            self.environment.hosts[1], router)
+        # ping external vm to test snat
+        vm.block_until_ping(external_vm.ip)
+
+        fip = self.safe_client.create_floatingip(
+            tenant_id, ext_net['id'], vm.ip, vm.neutron_port['id'])
+        # ping floating ip from external vm
+        external_vm.block_until_ping(fip['floating_ip_address'])
+
+        # ping router gateway IP
+        old_gw_ip = router['external_gateway_info'][
+            'external_fixed_ips'][0]['ip_address']
+        external_vm.block_until_ping(old_gw_ip)
+
+        gateway_port = self.safe_client.list_ports(
+            device_id=router['id'],
+            device_owner=constants.DEVICE_OWNER_ROUTER_GW)[0]
+        ip_1 = str(netaddr.IPNetwork(
+            ext_sub['gateway_ip']).next(100)).split('/')[0]
+        ip_2 = str(netaddr.IPNetwork(
+            ext_sub['gateway_ip']).next(101)).split('/')[0]
+        self.safe_client.update_port(gateway_port['id'], fixed_ips=[
+            {'ip_address': ip_1},
+            {'ip_address': ip_2}])
+        # ping router gateway new IPs
+        external_vm.block_until_ping(ip_1)
+        external_vm.block_until_ping(ip_2)
+
+        # ping router old gateway IP, should fail now
+        external_vm.block_until_no_ping(old_gw_ip)
+
 
 class TestLegacyL3Agent(TestL3Agent):
 
@@ -224,6 +269,9 @@ class TestLegacyL3Agent(TestL3Agent):
         # Verify north-south connectivity using ping6 to external_vm.
         vm.block_until_ping(external_vm.ipv6)
 
+    def test_gateway_ip_changed(self):
+        self._test_gateway_ip_changed()
+
 
 class TestHAL3Agent(TestL3Agent):
 
@@ -375,3 +423,6 @@ class TestHAL3Agent(TestL3Agent):
 
         self._assert_ping_during_agents_restart(
             l3_active_agents, external_vm.namespace, [router_ip], count=60)
+
+    def test_gateway_ip_changed(self):
+        self._test_gateway_ip_changed()
diff --git a/neutron/tests/unit/extensions/test_l3.py b/neutron/tests/unit/extensions/test_l3.py
index e98ccb5b33e..0f6e251b7ce 100644
--- a/neutron/tests/unit/extensions/test_l3.py
+++ b/neutron/tests/unit/extensions/test_l3.py
@@ -269,6 +269,21 @@ class TestL3NatBasePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             plugin.disassociate_floatingips(context, id)
         return super(TestL3NatBasePlugin, self).delete_port(context, id)
 
+    def update_port(self, context, id, port):
+        original_port = self.get_port(context, id)
+        session = context.session
+        with session.begin(subtransactions=True):
+            new_port = super(TestL3NatBasePlugin, self).update_port(
+                context, id, port)
+        # Notifications must be sent after the above transaction is complete
+        kwargs = {
+            'context': context,
+            'port': new_port,
+            'original_port': original_port,
+        }
+        registry.notify(resources.PORT, events.AFTER_UPDATE, self, **kwargs)
+        return new_port
+
 
 # This plugin class is for tests with plugin that integrates L3.
 class TestL3NatIntPlugin(TestL3NatBasePlugin,
@@ -3434,6 +3449,73 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
                     self.assertEqual('', body['port']['device_owner'])
                     self.assertEqual('', body['port']['device_id'])
 
+    def _test__notify_gateway_port_ip_changed_helper(self, gw_ip_change=True):
+        plugin = directory.get_plugin(plugin_constants.L3)
+        if not hasattr(plugin, 'l3_rpc_notifier'):
+            self.skipTest("Plugin does not support l3_rpc_notifier")
+        # make sure the callback is registered.
+        registry.subscribe(
+            l3_db.L3RpcNotifierMixin._notify_gateway_port_ip_changed,
+            resources.PORT,
+            events.AFTER_UPDATE)
+        with mock.patch.object(plugin.l3_rpc_notifier,
+                               'routers_updated') as chk_method:
+            with self.router() as router:
+                with self.subnet(cidr='1.1.1.0/24') as subnet:
+                    self._set_net_external(subnet['subnet']['network_id'])
+                    router_id = router['router']['id']
+                    self._add_external_gateway_to_router(
+                        router_id,
+                        subnet['subnet']['network_id'])
+                    body = self._show('routers', router_id)
+                    gateway_ips = body['router']['external_gateway_info'][
+                        'external_fixed_ips']
+                    gateway_ip_len = len(gateway_ips)
+                    self.assertEqual(1, gateway_ip_len)
+                    gw_port_id = None
+                    for p in self._list('ports')['ports']:
+                        if (p['device_owner'] ==
+                                lib_constants.DEVICE_OWNER_ROUTER_GW and
+                                p['device_id'] == router_id):
+                            gw_port_id = p['id']
+                    self.assertIsNotNone(gw_port_id)
+                    gw_ip_len = 1
+                    if gw_ip_change:
+                        gw_ip_len += 1
+                        data = {'port': {'fixed_ips': [
+                            {'ip_address': '1.1.1.101'},
+                            {'ip_address': '1.1.1.100'}]}}
+                    else:
+                        gw_ip = gateway_ips[0]['ip_address']
+                        data = {'port': {'fixed_ips': [
+                            {'ip_address': gw_ip}]}}
+                    req = self.new_update_request('ports', data,
+                                                  gw_port_id)
+                    res = self.deserialize(self.fmt,
+                                           req.get_response(self.api))
+                    self.assertEqual(gw_ip_len, len(res['port']['fixed_ips']))
+
+                    body = self._show('routers', router_id)
+                    gateway_ip_len = len(
+                        body['router']['external_gateway_info'][
+                            'external_fixed_ips'])
+                    self.assertEqual(gw_ip_len, gateway_ip_len)
+                    chk_method.assert_called_with(mock.ANY,
+                                                  [router_id], None)
+                    self.assertEqual(gw_ip_len, chk_method.call_count)
+
+    def test__notify_gateway_port_ip_changed(self):
+        """Test to make sure notification to routers occurs when the gateway
+            ip address changed.
+        """
+        self._test__notify_gateway_port_ip_changed_helper()
+
+    def test__notify_gateway_port_ip_not_changed(self):
+        """Test to make sure no notification to routers occurs when the gateway
+            ip address is not changed.
+        """
+        self._test__notify_gateway_port_ip_changed_helper(gw_ip_change=False)
+
     def test_update_subnet_gateway_for_external_net(self):
         """Test to make sure notification to routers occurs when the gateway
             ip address of a subnet of the external network is changed.