diff --git a/elasticsearch/_async/transport.py b/elasticsearch/_async/transport.py
index e5ed4b87..9fa94107 100644
--- a/elasticsearch/_async/transport.py
+++ b/elasticsearch/_async/transport.py
@@ -27,11 +27,10 @@ from ..exceptions import (
     ConnectionError,
     ConnectionTimeout,
     ElasticsearchWarning,
-    NotElasticsearchError,
     SerializationError,
     TransportError,
 )
-from ..transport import Transport, _verify_elasticsearch
+from ..transport import Transport, _ProductChecker
 from .compat import get_running_loop
 from .http_aiohttp import AIOHttpConnection
 
@@ -340,12 +339,9 @@ class AsyncTransport(Transport):
         if self._verified_elasticsearch is None:
             await self._do_verify_elasticsearch(headers=headers, timeout=timeout)
 
-        # If '_verified_elasticsearch' is False we know we're not connected to Elasticsearch.
-        if self._verified_elasticsearch is False:
-            raise NotElasticsearchError(
-                "The client noticed that the server is not Elasticsearch "
-                "and we do not support this unknown product"
-            )
+        # If '_verified_elasticsearch' isn't 'True' then we raise an error.
+        if self._verified_elasticsearch is not True:
+            _ProductChecker.raise_error(self._verified_elasticsearch)
 
         for attempt in range(self.max_retries + 1):
             connection = self.get_connection()
@@ -496,6 +492,6 @@ class AsyncTransport(Transport):
                 raise error
 
             # Check the information we got back from the index request.
-            self._verified_elasticsearch = _verify_elasticsearch(
+            self._verified_elasticsearch = _ProductChecker.check_product(
                 info_headers, info_response
             )
diff --git a/elasticsearch/exceptions.py b/elasticsearch/exceptions.py
index 2b3c5ec2..9ff5c8b2 100644
--- a/elasticsearch/exceptions.py
+++ b/elasticsearch/exceptions.py
@@ -51,9 +51,9 @@ class SerializationError(ElasticsearchException):
     """
 
 
-class NotElasticsearchError(ElasticsearchException):
+class UnsupportedProductError(ElasticsearchException):
     """Error which is raised when the client detects
-    it's not connected to an Elasticsearch cluster.
+    it's not connected to a supported product.
     """
 
 
diff --git a/elasticsearch/exceptions.pyi b/elasticsearch/exceptions.pyi
index 1dcab8c0..af39f2d5 100644
--- a/elasticsearch/exceptions.pyi
+++ b/elasticsearch/exceptions.pyi
@@ -20,7 +20,7 @@ from typing import Any, Dict, Union
 class ImproperlyConfigured(Exception): ...
 class ElasticsearchException(Exception): ...
 class SerializationError(ElasticsearchException): ...
-class NotElasticsearchError(ElasticsearchException): ...
+class UnsupportedProductError(ElasticsearchException): ...
 
 class TransportError(ElasticsearchException):
     @property
diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py
index 0ee68106..5e9d80e2 100644
--- a/elasticsearch/transport.py
+++ b/elasticsearch/transport.py
@@ -31,9 +31,9 @@ from .exceptions import (
     ConnectionError,
     ConnectionTimeout,
     ElasticsearchWarning,
-    NotElasticsearchError,
     SerializationError,
     TransportError,
+    UnsupportedProductError,
 )
 from .serializer import DEFAULT_SERIALIZERS, Deserializer, JSONSerializer
 from .utils import _client_meta_version
@@ -214,8 +214,8 @@ class Transport(object):
         # - 'True': Means we've verified that we're talking to Elasticsearch or
         #   that we can't rule out Elasticsearch due to auth issues. A warning
         #   will be raised if we receive 401/403.
-        # - 'False': Means we've discovered we're not talking to Elasticsearch,
-        #   should raise an error in this case for every request.
+        # - 'int': Means we're talking to an unsupported product, should raise
+        #   the corresponding error.
         self._verified_elasticsearch = None
 
         # Ensures that the ES verification request only fires once and that
@@ -408,12 +408,9 @@ class Transport(object):
         if self._verified_elasticsearch is None:
             self._do_verify_elasticsearch(headers=headers, timeout=timeout)
 
-        # If '_verified_elasticsearch' is False we know we're not connected to Elasticsearch.
-        if self._verified_elasticsearch is False:
-            raise NotElasticsearchError(
-                "The client noticed that the server is not Elasticsearch "
-                "and we do not support this unknown product"
-            )
+        # If '_verified_elasticsearch' isn't 'True' then we raise an error.
+        if self._verified_elasticsearch is not True:
+            _ProductChecker.raise_error(self._verified_elasticsearch)
 
         for attempt in range(self.max_retries + 1):
             connection = self.get_connection()
@@ -601,53 +598,84 @@ class Transport(object):
                 raise error
 
             # Check the information we got back from the index request.
-            self._verified_elasticsearch = _verify_elasticsearch(
+            self._verified_elasticsearch = _ProductChecker.check_product(
                 info_headers, info_response
             )
 
 
-def _verify_elasticsearch(headers, response):
-    """Verifies that the server we're talking to is Elasticsearch.
-    Does this by checking HTTP headers and the deserialized
-    response to the 'info' API. Returns 'True' if we're verified
-    against Elasticsearch, 'False' otherwise.
-    """
-    try:
-        version = response.get("version", {})
-        version_number = tuple(
-            int(x) if x is not None else 999
-            for x in re.search(
-                r"^([0-9]+)\.([0-9]+)(?:\.([0-9]+))?", version["number"]
-            ).groups()
-        )
-    except (KeyError, TypeError, ValueError, AttributeError):
-        # No valid 'version.number' field, effectively 0.0.0
-        version = {}
-        version_number = (0, 0, 0)
+class _ProductChecker:
+    """Class which verifies we're connected to a supported product"""
 
-    # Check all of the fields and headers for missing/valid values.
-    try:
-        bad_tagline = response.get("tagline", None) != "You Know, for Search"
-        bad_build_flavor = version.get("build_flavor", None) != "default"
-        bad_product_header = headers.get("x-elastic-product", None) != "Elasticsearch"
-    except (AttributeError, TypeError):
-        bad_tagline = True
-        bad_build_flavor = True
-        bad_product_header = True
+    # States that can be returned from 'check_product'
+    SUCCESS = True
+    UNSUPPORTED_PRODUCT = 2
+    UNSUPPORTED_DISTRIBUTION = 3
 
-    if (
-        # No version or version less than 6.x
-        version_number < (6, 0, 0)
-        # 6.x and there's a bad 'tagline'
-        or ((6, 0, 0) <= version_number < (7, 0, 0) and bad_tagline)
-        # 7.0-7.13 and there's a bad 'tagline' or 'build_flavor'
-        or (
-            (7, 0, 0) <= version_number < (7, 14, 0)
-            and (bad_tagline or bad_build_flavor)
-        )
-        # 7.14+ and there's a bad 'X-Elastic-Product' HTTP header
-        or ((7, 14, 0) <= version_number and bad_product_header)
-    ):
-        return False
+    @classmethod
+    def raise_error(cls, state):
+        # These states mean the product_check() didn't fail so do nothing.
+        if state in (None, True):
+            return
 
-    return True
+        if state == cls.UNSUPPORTED_DISTRIBUTION:
+            message = (
+                "The client noticed that the server is not "
+                "a supported distribution of Elasticsearch"
+            )
+        else:  # UNSUPPORTED_PRODUCT
+            message = (
+                "The client noticed that the server is not Elasticsearch "
+                "and we do not support this unknown product"
+            )
+        raise UnsupportedProductError(message)
+
+    @classmethod
+    def check_product(cls, headers, response):
+        # type: (dict[str, str], dict[str, str]) -> int
+        """Verifies that the server we're talking to is Elasticsearch.
+        Does this by checking HTTP headers and the deserialized
+        response to the 'info' API. Returns one of the states above.
+        """
+        try:
+            version = response.get("version", {})
+            version_number = tuple(
+                int(x) if x is not None else 999
+                for x in re.search(
+                    r"^([0-9]+)\.([0-9]+)(?:\.([0-9]+))?", version["number"]
+                ).groups()
+            )
+        except (KeyError, TypeError, ValueError, AttributeError):
+            # No valid 'version.number' field, effectively 0.0.0
+            version = {}
+            version_number = (0, 0, 0)
+
+        # Check all of the fields and headers for missing/valid values.
+        try:
+            bad_tagline = response.get("tagline", None) != "You Know, for Search"
+            bad_build_flavor = version.get("build_flavor", None) != "default"
+            bad_product_header = (
+                headers.get("x-elastic-product", None) != "Elasticsearch"
+            )
+        except (AttributeError, TypeError):
+            bad_tagline = True
+            bad_build_flavor = True
+            bad_product_header = True
+
+        # 7.0-7.13 and there's a bad 'tagline' or unsupported 'build_flavor'
+        if (7, 0, 0) <= version_number < (7, 14, 0):
+            if bad_tagline:
+                return cls.UNSUPPORTED_PRODUCT
+            elif bad_build_flavor:
+                return cls.UNSUPPORTED_DISTRIBUTION
+
+        elif (
+            # No version or version less than 6.x
+            version_number < (6, 0, 0)
+            # 6.x and there's a bad 'tagline'
+            or ((6, 0, 0) <= version_number < (7, 0, 0) and bad_tagline)
+            # 7.14+ and there's a bad 'X-Elastic-Product' HTTP header
+            or ((7, 14, 0) <= version_number and bad_product_header)
+        ):
+            return cls.UNSUPPORTED_PRODUCT
+
+        return True
diff --git a/test_elasticsearch/test_async/test_transport.py b/test_elasticsearch/test_async/test_transport.py
index fb93fefa..9b3743d7 100644
--- a/test_elasticsearch/test_async/test_transport.py
+++ b/test_elasticsearch/test_async/test_transport.py
@@ -33,10 +33,11 @@ from elasticsearch.exceptions import (
     AuthorizationException,
     ConnectionError,
     ElasticsearchWarning,
-    NotElasticsearchError,
     NotFoundError,
     TransportError,
+    UnsupportedProductError,
 )
+from elasticsearch.transport import _ProductChecker
 
 pytestmark = pytest.mark.asyncio
 
@@ -641,7 +642,7 @@ class TestTransport:
             [{"data": data, "headers": headers}], connection_class=DummyConnection
         )
         await t.perform_request("GET", "/_search")
-        assert t._verified_elasticsearch
+        assert t._verified_elasticsearch is True
 
         calls = t.connection_pool.connections[0].calls
         _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls]
@@ -688,7 +689,7 @@ class TestTransport:
         ]
 
         # Assert that the cluster is "verified"
-        assert t._verified_elasticsearch
+        assert t._verified_elasticsearch is True
 
         # See that the headers were passed along to the "info" request made
         calls = t.connection_pool.connections[0].calls
@@ -762,7 +763,7 @@ class TestTransport:
         )
 
         # Assert that the cluster is "verified"
-        assert t._verified_elasticsearch
+        assert t._verified_elasticsearch is True
 
         # See that the first request is always 'GET /' for ES check
         calls = t.connection_pool.connections[0].calls
@@ -771,13 +772,37 @@ class TestTransport:
         # The rest of the requests are 'GET /_search' afterwards
         assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
 
+    @pytest.mark.parametrize(
+        ["build_flavor", "tagline", "product_error", "error_message"],
+        [
+            (
+                "default",
+                "BAD TAGLINE",
+                _ProductChecker.UNSUPPORTED_PRODUCT,
+                "The client noticed that the server is not Elasticsearch and we do not support this unknown product",
+            ),
+            (
+                "BAD BUILD FLAVOR",
+                "BAD TAGLINE",
+                _ProductChecker.UNSUPPORTED_PRODUCT,
+                "The client noticed that the server is not Elasticsearch and we do not support this unknown product",
+            ),
+            (
+                "BAD BUILD FLAVOR",
+                "You Know, for Search",
+                _ProductChecker.UNSUPPORTED_DISTRIBUTION,
+                "The client noticed that the server is not a supported distribution of Elasticsearch",
+            ),
+        ],
+    )
     async def test_multiple_requests_verify_elasticsearch_product_error(
-        self, event_loop
+        self, event_loop, build_flavor, tagline, product_error, error_message
     ):
         t = AsyncTransport(
             [
                 {
-                    "data": '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"BAD TAGLINE"}',
+                    "data": '{"version":{"number":"7.13.0","build_flavor":"%s"},"tagline":"%s"}'
+                    % (build_flavor, tagline),
                     "delay": 1,
                 }
             ],
@@ -806,7 +831,8 @@ class TestTransport:
         assert len(results) == 10
 
         # All results were errors
-        assert all(isinstance(result, NotElasticsearchError) for result in results)
+        assert all(isinstance(result, UnsupportedProductError) for result in results)
+        assert all(str(result) == error_message for result in results)
 
         # Assert that one request was made but not 2 requests.
         duration = end_time - start_time
@@ -818,7 +844,7 @@ class TestTransport:
         )
 
         # Assert that the cluster is definitely not Elasticsearch
-        assert t._verified_elasticsearch is False
+        assert t._verified_elasticsearch == product_error
 
         # See that the first request is always 'GET /' for ES check
         calls = t.connection_pool.connections[0].calls
diff --git a/test_elasticsearch/test_transport.py b/test_elasticsearch/test_transport.py
index 84ac388b..a9a92401 100644
--- a/test_elasticsearch/test_transport.py
+++ b/test_elasticsearch/test_transport.py
@@ -31,13 +31,11 @@ from elasticsearch.exceptions import (
     AuthorizationException,
     ConnectionError,
     ElasticsearchWarning,
-    NotElasticsearchError,
     NotFoundError,
     TransportError,
+    UnsupportedProductError,
 )
-from elasticsearch.transport import Transport
-from elasticsearch.transport import _verify_elasticsearch as verify_elasticsearch
-from elasticsearch.transport import get_host_info
+from elasticsearch.transport import Transport, _ProductChecker, get_host_info
 
 from .test_cases import TestCase
 
@@ -486,26 +484,50 @@ TAGLINE = "You Know, for Search"
 
 
 @pytest.mark.parametrize(
-    ["headers", "response"],
+    ["headers", "response", "product_error"],
     [
         # All empty.
-        ({}, {}),
+        ({}, {}, _ProductChecker.UNSUPPORTED_PRODUCT),
         # Don't check the product header immediately, need to check version first.
-        ({"x-elastic-product": "Elasticsearch"}, {}),
+        (
+            {"x-elastic-product": "Elasticsearch"},
+            {},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Version not there.
-        ({}, {"tagline": TAGLINE}),
+        ({}, {"tagline": TAGLINE}, _ProductChecker.UNSUPPORTED_PRODUCT),
         # Version is nonsense
-        ({}, {"version": "1.0.0", "tagline": TAGLINE}),
+        (
+            {},
+            {"version": "1.0.0", "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Version number not there
-        ({}, {"version": {}, "tagline": TAGLINE}),
+        ({}, {"version": {}, "tagline": TAGLINE}, _ProductChecker.UNSUPPORTED_PRODUCT),
         # Version number is nonsense
-        ({}, {"version": {"number": "nonsense"}, "tagline": TAGLINE}),
+        (
+            {},
+            {"version": {"number": "nonsense"}, "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Version number way in the past
-        ({}, {"version": {"number": "1.0.0"}, "tagline": TAGLINE}),
+        (
+            {},
+            {"version": {"number": "1.0.0"}, "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Version number way in the future
-        ({}, {"version": {"number": "999.0.0"}, "tagline": TAGLINE}),
+        (
+            {},
+            {"version": {"number": "999.0.0"}, "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Build flavor not supposed to be missing
-        ({}, {"version": {"number": "7.13.0"}, "tagline": TAGLINE}),
+        (
+            {},
+            {"version": {"number": "7.13.0"}, "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_DISTRIBUTION,
+        ),
         # Build flavor is 'oss'
         (
             {},
@@ -513,6 +535,7 @@ TAGLINE = "You Know, for Search"
                 "version": {"number": "7.10.0", "build_flavor": "oss"},
                 "tagline": TAGLINE,
             },
+            _ProductChecker.UNSUPPORTED_DISTRIBUTION,
         ),
         # Build flavor is nonsense
         (
@@ -521,20 +544,30 @@ TAGLINE = "You Know, for Search"
                 "version": {"number": "7.13.0", "build_flavor": "nonsense"},
                 "tagline": TAGLINE,
             },
+            _ProductChecker.UNSUPPORTED_DISTRIBUTION,
         ),
         # Tagline is nonsense
-        ({}, {"version": {"number": "7.1.0-SNAPSHOT"}, "tagline": "nonsense"}),
+        (
+            {},
+            {"version": {"number": "7.1.0-SNAPSHOT"}, "tagline": "nonsense"},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Product header is not supposed to be missing
-        ({}, {"version": {"number": "7.14.0"}, "tagline": "You Know, for Search"}),
+        (
+            {},
+            {"version": {"number": "7.14.0"}, "tagline": "You Know, for Search"},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+        ),
         # Product header is nonsense
         (
             {"x-elastic-product": "nonsense"},
             {"version": {"number": "7.15.0"}, "tagline": TAGLINE},
+            _ProductChecker.UNSUPPORTED_PRODUCT,
         ),
     ],
 )
-def test_verify_elasticsearch_errors(headers, response):
-    assert verify_elasticsearch(headers, response) is False
+def test_verify_elasticsearch_errors(headers, response, product_error):
+    assert _ProductChecker.check_product(headers, response) == product_error
 
 
 @pytest.mark.parametrize(
@@ -579,7 +612,9 @@ def test_verify_elasticsearch_errors(headers, response):
     ],
 )
 def test_verify_elasticsearch_passes(headers, response):
-    assert verify_elasticsearch(headers, response) is True
+    result = _ProductChecker.check_product(headers, response)
+    assert result == _ProductChecker.SUCCESS
+    assert result is True
 
 
 @pytest.mark.parametrize(
@@ -624,7 +659,7 @@ def test_verify_elasticsearch(headers, data):
         [{"data": data, "headers": headers}], connection_class=DummyConnection
     )
     t.perform_request("GET", "/_search")
-    assert t._verified_elasticsearch
+    assert t._verified_elasticsearch is True
 
     calls = t.connection_pool.connections[0].calls
     _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls]
@@ -675,7 +710,7 @@ def test_verify_elasticsearch_skips_on_auth_errors(exception_cls):
     ]
 
     # Assert that the cluster is "verified"
-    assert t._verified_elasticsearch
+    assert t._verified_elasticsearch is True
 
     # See that the headers were passed along to the "info" request made
     calls = t.connection_pool.connections[0].calls
@@ -759,7 +794,7 @@ def test_multiple_requests_verify_elasticsearch_success():
     )
 
     # Assert that the cluster is "verified"
-    assert t._verified_elasticsearch
+    assert t._verified_elasticsearch is True
 
     # See that the first request is always 'GET /' for ES check
     calls = t.connection_pool.connections[0].calls
@@ -769,7 +804,32 @@ def test_multiple_requests_verify_elasticsearch_success():
     assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
 
 
-def test_multiple_requests_verify_elasticsearch_product_error():
+@pytest.mark.parametrize(
+    ["build_flavor", "tagline", "product_error", "error_message"],
+    [
+        (
+            "default",
+            "BAD TAGLINE",
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+            "The client noticed that the server is not Elasticsearch and we do not support this unknown product",
+        ),
+        (
+            "BAD BUILD FLAVOR",
+            "BAD TAGLINE",
+            _ProductChecker.UNSUPPORTED_PRODUCT,
+            "The client noticed that the server is not Elasticsearch and we do not support this unknown product",
+        ),
+        (
+            "BAD BUILD FLAVOR",
+            "You Know, for Search",
+            _ProductChecker.UNSUPPORTED_DISTRIBUTION,
+            "The client noticed that the server is not a supported distribution of Elasticsearch",
+        ),
+    ],
+)
+def test_multiple_requests_verify_elasticsearch_product_error(
+    build_flavor, tagline, product_error, error_message
+):
     try:
         import threading
     except ImportError:
@@ -778,7 +838,8 @@ def test_multiple_requests_verify_elasticsearch_product_error():
     t = Transport(
         [
             {
-                "data": '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"BAD TAGLINE"}',
+                "data": '{"version":{"number":"7.13.0","build_flavor":"%s"},"tagline":"%s"}'
+                % (build_flavor, tagline),
                 "delay": 1,
             }
         ],
@@ -811,7 +872,8 @@ def test_multiple_requests_verify_elasticsearch_product_error():
     assert len(results) == 10
 
     # All results were errors
-    assert all(isinstance(result, NotElasticsearchError) for result in results)
+    assert all(isinstance(result, UnsupportedProductError) for result in results)
+    assert all(str(result) == error_message for result in results)
 
     # Assert that one request was made but not 2 requests.
     duration = end_time - start_time
@@ -823,7 +885,7 @@ def test_multiple_requests_verify_elasticsearch_product_error():
     )
 
     # Assert that the cluster is definitely not Elasticsearch
-    assert t._verified_elasticsearch is False
+    assert t._verified_elasticsearch == product_error
 
     # See that the first request is always 'GET /' for ES check
     calls = t.connection_pool.connections[0].calls
@@ -875,7 +937,7 @@ def test_multiple_requests_verify_elasticsearch_retry_on_errors(error_cls):
     # Exactly 5 results completed
     assert len(results) == 5
 
-    # All results were errors and not wrapped in 'NotElasticsearchError'
+    # All results were errors and not wrapped in 'UnsupportedProductError'
     assert all(isinstance(result, error_cls) for result in results)
 
     # Assert that 5 requests were made in total (5 transport requests per x 0.1s/conn request)