diff --git a/neutron/db/l3_db.py b/neutron/db/l3_db.py
index e11719647c5..52777a4cb4e 100644
--- a/neutron/db/l3_db.py
+++ b/neutron/db/l3_db.py
@@ -1252,6 +1252,13 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
             context, internal_port,
             internal_subnet_id, floatingip_obj.floating_network_id)
 
+        if self.is_router_distributed(context, router_id):
+            if not plugin_utils.can_port_be_bound_to_virtual_bridge(
+                    internal_port):
+                msg = _('Port VNIC type is not valid to associate a FIP in '
+                        'DVR mode')
+                raise n_exc.BadRequest(resource='floatingip', msg=msg)
+
         return (fip['port_id'], internal_ip_address, router_id)
 
     def _check_and_get_fip_assoc(self, context, fip, floatingip_obj):
@@ -1874,6 +1881,15 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
         self._process_interfaces(routers_dict, interfaces)
         return list(routers_dict.values())
 
+    def is_router_distributed(self, context, router_id):
+        """Returns if a router is distributed or not
+
+        If DVR extension is not enabled, no router will be distributed. This
+        function is overridden in L3_NAT_with_dvr_db_mixin in case the DVR
+        extension is loaded.
+        """
+        return False
+
 
 @registry.has_registry_receivers
 class L3RpcNotifierMixin(object):
diff --git a/neutron/db/l3_dvr_db.py b/neutron/db/l3_dvr_db.py
index 11a8af2a863..360320e07f7 100644
--- a/neutron/db/l3_dvr_db.py
+++ b/neutron/db/l3_dvr_db.py
@@ -1214,6 +1214,13 @@ class L3_NAT_with_dvr_db_mixin(_DVRAgentInterfaceMixin,
         floating_ip = self._delete_floatingip(context, id)
         self._notify_floating_ip_change(context, floating_ip)
 
+    @db_api.retry_if_session_inactive()
+    def is_router_distributed(self, context, router_id):
+        if router_id:
+            return is_distributed_router(
+                self.get_router(context.elevated(), router_id))
+        return False
+
 
 def is_distributed_router(router):
     """Return True if router to be handled is distributed."""
diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py
index 58181d41752..9ae3a0d7f94 100644
--- a/neutron/tests/unit/db/test_l3_dvr_db.py
+++ b/neutron/tests/unit/db/test_l3_dvr_db.py
@@ -24,6 +24,7 @@ from neutron_lib import exceptions
 from neutron_lib.exceptions import l3 as l3_exc
 from neutron_lib.plugins import constants as plugin_constants
 from neutron_lib.plugins import directory
+from neutron_lib.plugins import utils as plugin_utils
 from oslo_utils import uuidutils
 
 from neutron.db import agents_db
@@ -1047,3 +1048,44 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
                 routers = self.mixin._get_sync_routers(
                     self.ctx, router_ids=[router['id']])
                 self.assertEqual("fake-host", routers[0]['gw_port_host'])
+
+    def test_is_router_distributed(self):
+        router_id = 'router_id'
+        with mock.patch.object(self.mixin, 'get_router') as \
+                mock_get_router:
+            mock_get_router.return_value = {'distributed': True}
+            self.assertTrue(
+                self.mixin.is_router_distributed(self.ctx, router_id))
+
+    @mock.patch.object(plugin_utils, 'can_port_be_bound_to_virtual_bridge',
+                       return_value=True)
+    def test__get_assoc_data_valid_vnic_type(self, *args):
+        with mock.patch.object(self.mixin, '_internal_fip_assoc_data') as \
+                mock_fip_assoc_data, \
+                mock.patch.object(self.mixin, '_get_router_for_floatingip') \
+                as mock_router_fip, \
+                mock.patch.object(self.mixin, 'is_router_distributed',
+                                  return_value=True):
+            port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL}
+            mock_fip_assoc_data.return_value = (port, 'subnet_id', 'ip_addr')
+            mock_router_fip.return_value = 'router_id'
+            fip = {'port_id': 'port_id'}
+            self.assertEqual(
+                ('port_id', 'ip_addr', 'router_id'),
+                self.mixin._get_assoc_data(self.ctx, fip, mock.Mock()))
+
+    @mock.patch.object(plugin_utils, 'can_port_be_bound_to_virtual_bridge',
+                       return_value=False)
+    def test__get_assoc_data_invalid_vnic_type(self, *args):
+        with mock.patch.object(self.mixin, '_internal_fip_assoc_data') as \
+                mock_fip_assoc_data, \
+                mock.patch.object(self.mixin, '_get_router_for_floatingip') \
+                as mock_router_fip, \
+                mock.patch.object(self.mixin, 'is_router_distributed',
+                                  return_value=True):
+            port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL}
+            mock_fip_assoc_data.return_value = (port, 'subnet_id', 'ip_addr')
+            mock_router_fip.return_value = 'router_id'
+            self.assertRaises(
+                exceptions.BadRequest,
+                self.mixin._get_assoc_data, self.ctx, mock.ANY, mock.Mock())