diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md
index 88216cd2235f..66ba7b13875b 100644
--- a/sdk/cosmos/azure-cosmos/CHANGELOG.md
+++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md
@@ -3,6 +3,7 @@
 ### 4.3.0b5 (Unreleased)
 
 #### Other Changes
+- Deprecated offer-named methods in favor of their new throughput-named counterparts (`read_offer` -> `get_throughput`).
 - Marked the GetAuthorizationMethod for deprecation since it will no longer be public in a future release.
 - Added samples showing how to configure retry options for both the sync and async clients.
 - Deprecated the `connection_retry_policy` and `retry_options` options in the sync client.
@@ -31,7 +32,7 @@
 - Added new **provisional** `max_integrated_cache_staleness_in_ms` parameter to read item and query items APIs in order
   to make use of the **preview** CosmosDB integrated cache functionality.
   Please see [Azure Cosmos DB integrated cache](https://docs.microsoft.com/azure/cosmos-db/integrated-cache) for more details.
-- Added support for split-proof queries for the async client
+- Added support for split-proof queries for the async client.
 
 ### Bugs fixed
 - Default consistency level for the sync and async clients is no longer `Session` and will instead be set to the 
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/container.py
index 50c47c4026fd..fe6ec8b7a8b3 100644
--- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/container.py
+++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/container.py
@@ -32,7 +32,7 @@
 from .._base import build_options as _build_options, validate_cache_staleness_value
 from ..exceptions import CosmosResourceNotFoundError
 from ..http_constants import StatusCodes
-from ..offer import Offer
+from ..offer import ThroughputProperties
 from .scripts import ScriptsProxy
 from ..partition_key import NonePartitionKeyValue
 
@@ -114,10 +114,10 @@ def _set_partition_key(self, partition_key) -> Union[str, Awaitable]:
 
     @distributed_trace_async
     async def read(
-            self,
-            populate_partition_key_range_statistics=None,  # type: Optional[bool]
-            populate_quota_info=None,  # type: Optional[bool]
-            **kwargs  # type: Any
+        self,
+        populate_partition_key_range_statistics=None,  # type: Optional[bool]
+        populate_quota_info=None,  # type: Optional[bool]
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Read the container properties.
 
@@ -151,9 +151,9 @@ async def read(
 
     @distributed_trace_async
     async def create_item(
-            self,
-            body,  # type: Dict[str, Any]
-            **kwargs  # type: Any
+        self,
+        body,  # type: Dict[str, Any]
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Create an item in the container.
 
@@ -198,10 +198,10 @@ async def create_item(
 
     @distributed_trace_async
     async def read_item(
-            self,
-            item,  # type: Union[str, Dict[str, Any]]
-            partition_key,  # type: Any
-            **kwargs  # type: Any
+        self,
+        item,  # type: Union[str, Dict[str, Any]]
+        partition_key,  # type: Any
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Get the item identified by `item`.
 
@@ -245,9 +245,9 @@ async def read_item(
 
     @distributed_trace
     def read_all_items(
-            self,
-            max_item_count=None,  # type: Optional[int]
-            **kwargs  # type: Any
+        self,
+        max_item_count=None,  # type: Optional[int]
+        **kwargs  # type: Any
     ) -> AsyncItemPaged[Dict[str, Any]]:
         """List all the items in the container.
 
@@ -284,14 +284,14 @@ def read_all_items(
 
     @distributed_trace
     def query_items(
-            self,
-            query,  # type: str
-            parameters=None,  # type: Optional[List[Dict[str, Any]]]
-            partition_key=None,  # type: Optional[Any]
-            max_item_count=None,  # type: Optional[int]
-            enable_scan_in_query=None,  # type: Optional[bool]
-            populate_query_metrics=None,  # type: Optional[bool]
-            **kwargs  # type: Any
+        self,
+        query,  # type: str
+        parameters=None,  # type: Optional[List[Dict[str, Any]]]
+        partition_key=None,  # type: Optional[Any]
+        max_item_count=None,  # type: Optional[int]
+        enable_scan_in_query=None,  # type: Optional[bool]
+        populate_query_metrics=None,  # type: Optional[bool]
+        **kwargs  # type: Any
     ) -> AsyncItemPaged[Dict[str, Any]]:
         """Return all results matching the given `query`.
 
@@ -373,12 +373,12 @@ def query_items(
 
     @distributed_trace
     def query_items_change_feed(
-            self,
-            partition_key_range_id=None,  # type: Optional[str]
-            is_start_from_beginning=False,  # type: bool
-            continuation=None,  # type: Optional[str]
-            max_item_count=None,  # type: Optional[int]
-            **kwargs  # type: Any
+        self,
+        partition_key_range_id=None,  # type: Optional[str]
+        is_start_from_beginning=False,  # type: bool
+        continuation=None,  # type: Optional[str]
+        max_item_count=None,  # type: Optional[int]
+        **kwargs  # type: Any
     ) -> AsyncItemPaged[Dict[str, Any]]:
         """Get a sorted list of items that were changed, in the order in which they were modified.
 
@@ -419,11 +419,11 @@ def query_items_change_feed(
 
     @distributed_trace_async
     async def upsert_item(
-            self,
-            body,  # type: Dict[str, Any]
-            pre_trigger_include=None,  # type: Optional[str]
-            post_trigger_include=None,  # type: Optional[str]
-            **kwargs  # type: Any
+        self,
+        body,  # type: Dict[str, Any]
+        pre_trigger_include=None,  # type: Optional[str]
+        post_trigger_include=None,  # type: Optional[str]
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Insert or update the specified item.
 
@@ -463,12 +463,12 @@ async def upsert_item(
 
     @distributed_trace_async
     async def replace_item(
-            self,
-            item,  # type: Union[str, Dict[str, Any]]
-            body,  # type: Dict[str, Any]
-            pre_trigger_include=None,  # type: Optional[str]
-            post_trigger_include=None,  # type: Optional[str]
-            **kwargs  # type: Any
+        self,
+        item,  # type: Union[str, Dict[str, Any]]
+        body,  # type: Dict[str, Any]
+        pre_trigger_include=None,  # type: Optional[str]
+        post_trigger_include=None,  # type: Optional[str]
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Replaces the specified item if it exists in the container.
 
@@ -507,12 +507,12 @@ async def replace_item(
 
     @distributed_trace_async
     async def delete_item(
-            self,
-            item,  # type: Union[str, Dict[str, Any]]
-            partition_key,  # type: Any
-            pre_trigger_include=None,  # type: Optional[str]
-            post_trigger_include=None,  # type: Optional[str]
-            **kwargs  # type: Any
+        self,
+        item,  # type: Union[str, Dict[str, Any]]
+        partition_key,  # type: Any
+        pre_trigger_include=None,  # type: Optional[str]
+        post_trigger_include=None,  # type: Optional[str]
+        **kwargs  # type: Any
     ) -> None:
         """Delete the specified item from the container.
 
@@ -546,17 +546,17 @@ async def delete_item(
             response_hook(self.client_connection.last_response_headers, result)
 
     @distributed_trace_async
-    async def read_offer(self, **kwargs):
-        # type: (Any) -> Offer
-        """Read the Offer object for this container.
+    async def get_throughput(self, **kwargs):
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this container.
 
-        If no Offer already exists for the container, an exception is raised.
+        If no ThroughputProperties already exist for the container, an exception is raised.
 
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the container.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No offer exists for the container or
-            the offer could not be retrieved.
-        :rtype: ~azure.cosmos.Offer
+        :returns: ThroughputProperties for the container.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = await self._get_properties()
@@ -565,30 +565,33 @@ async def read_offer(self, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = [offer async for offer in self.client_connection.QueryOffers(query_spec, **kwargs)]
-        if len(offers) == 0:
+        throughput_properties = [throughput async for throughput in
+                                 self.client_connection.QueryOffers(query_spec, **kwargs)]
+        if len(throughput_properties) == 0:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for database " + self.database_link)
+                message="Could not find ThroughputProperties for container " + self.container_link)
 
         if response_hook:
-            response_hook(self.client_connection.last_response_headers, offers)
+            response_hook(self.client_connection.last_response_headers, throughput_properties)
 
-        return Offer(offer_throughput=offers[0]["content"]["offerThroughput"], properties=offers[0])
+        return ThroughputProperties(
+            offer_throughput=throughput_properties[0]["content"]["offerThroughput"],
+            properties=throughput_properties[0])
 
     @distributed_trace_async
     async def replace_throughput(self, throughput, **kwargs):
-        # type: (int, Any) -> Offer
+        # type: (int, Any) -> ThroughputProperties
         """Replace the container's throughput.
 
-        If no Offer already exists for the container, an exception is raised.
+        If no ThroughputProperties already exist for the container, an exception is raised.
 
         :param throughput: The throughput to be set (an integer).
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the container, updated with new throughput.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No offer exists for the container
-            or the offer could not be updated.
-        :rtype: ~azure.cosmos.Offer
+        :returns: ThroughputProperties for the container, updated with new throughput.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exist for the container
+            or the throughput properties could not be updated.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = await self._get_properties()
@@ -597,19 +600,21 @@ async def replace_throughput(self, throughput, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = [offer async for offer in self.client_connection.QueryOffers(query_spec, **kwargs)]
-        if len(offers) == 0:
+        throughput_properties = [throughput async for throughput in
+                                 self.client_connection.QueryOffers(query_spec, **kwargs)]
+        if len(throughput_properties) == 0:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for database " + self.database_link)
+                message="Could not find ThroughputProperties for container " + self.container_link)
 
-        new_offer = offers[0].copy()
-        new_offer["content"]["offerThroughput"] = throughput
-        data = await self.client_connection.ReplaceOffer(offer_link=offers[0]["_self"], offer=offers[0], **kwargs)
+        new_throughput_properties = throughput_properties[0].copy()
+        new_throughput_properties["content"]["offerThroughput"] = throughput
+        data = await self.client_connection.ReplaceOffer(offer_link=throughput_properties[0]["_self"],
+                                                         offer=throughput_properties[0], **kwargs)
         if response_hook:
             response_hook(self.client_connection.last_response_headers, data)
 
-        return Offer(offer_throughput=data["content"]["offerThroughput"], properties=data)
+        return ThroughputProperties(offer_throughput=data["content"]["offerThroughput"], properties=data)
 
     @distributed_trace
     def list_conflicts(self, max_item_count=None, **kwargs):
@@ -635,12 +640,12 @@ def list_conflicts(self, max_item_count=None, **kwargs):
 
     @distributed_trace
     def query_conflicts(
-            self,
-            query,  # type: str
-            parameters=None,  # type: Optional[List[Dict[str, Any]]]
-            partition_key=None,  # type: Optional[Any]
-            max_item_count=None,  # type: Optional[int]
-            **kwargs  # type: Any
+        self,
+        query,  # type: str
+        parameters=None,  # type: Optional[List[Dict[str, Any]]]
+        partition_key=None,  # type: Optional[Any]
+        max_item_count=None,  # type: Optional[int]
+        **kwargs  # type: Any
     ) -> AsyncItemPaged[Dict[str, Any]]:
         """Return all conflicts matching a given `query`.
 
@@ -673,11 +678,11 @@ def query_conflicts(
         return result
 
     @distributed_trace_async
-    async def read_conflict(
-            self,
-            conflict,  # type: Union[str, Dict[str, Any]]
-            partition_key,  # type: Any
-            **kwargs  # type: Any
+    async def get_conflict(
+        self,
+        conflict,  # type: Union[str, Dict[str, Any]]
+        partition_key,  # type: Any
+        **kwargs  # type: Any
     ) -> Dict[str, Any]:
         """Get the conflict identified by `conflict`.
 
@@ -700,10 +705,10 @@ async def read_conflict(
 
     @distributed_trace_async
     async def delete_conflict(
-            self,
-            conflict,  # type: Union[str, Dict[str, Any]]
-            partition_key,  # type: Any
-            **kwargs  # type: Any
+        self,
+        conflict,  # type: Union[str, Dict[str, Any]]
+        partition_key,  # type: Any
+        **kwargs  # type: Any
     ) -> None:
         """Delete a specified conflict from the container.
 
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/database.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/database.py
index 9561dae99192..887af36a7984 100644
--- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/database.py
+++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/database.py
@@ -32,7 +32,7 @@
 from ._cosmos_client_connection_async import CosmosClientConnection
 from .._base import build_options as _build_options
 from .container import ContainerProxy
-from ..offer import Offer
+from ..offer import ThroughputProperties
 from ..http_constants import StatusCodes
 from ..exceptions import CosmosResourceNotFoundError
 from .user import UserProxy
@@ -40,6 +40,7 @@
 
 __all__ = ("DatabaseProxy",)
 
+
 # pylint: disable=protected-access
 # pylint: disable=missing-client-constructor-parameter-credential,missing-client-constructor-parameter-kwargs
 
@@ -685,16 +686,19 @@ async def delete_user(self, user, **kwargs):
         if response_hook:
             response_hook(self.client_connection.last_response_headers, result)
 
+
     @distributed_trace_async
-    async def read_offer(self, **kwargs):
-        # type: (Any) -> Offer
-        """Read the Offer object for this database.
+    async def get_throughput(self, **kwargs):
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this database.
+
+        If no ThroughputProperties already exist for the database, an exception is raised.
 
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the database.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError:
-            If no offer exists for the database or if the offer could not be retrieved.
-        :rtype: ~azure.cosmos.Offer
+        :returns: ThroughputProperties for the database.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = await self._get_properties()
@@ -703,28 +707,30 @@ async def read_offer(self, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = [offer async for offer in self.client_connection.QueryOffers(query_spec, **kwargs)]
-        if len(offers) == 0:
+        throughput_properties = [throughput async for throughput in
+                                 self.client_connection.QueryOffers(query_spec, **kwargs)]
+        if len(throughput_properties) == 0:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for database " + self.database_link)
+                message="Could not find ThroughputProperties for database " + self.database_link)
 
         if response_hook:
-            response_hook(self.client_connection.last_response_headers, offers)
+            response_hook(self.client_connection.last_response_headers, throughput_properties)
 
-        return Offer(offer_throughput=offers[0]["content"]["offerThroughput"], properties=offers[0])
+        return ThroughputProperties(offer_throughput=throughput_properties[0]["content"]["offerThroughput"],
+                                    properties=throughput_properties[0])
 
     @distributed_trace_async
     async def replace_throughput(self, throughput, **kwargs):
-        # type: (Optional[int], Any) -> Offer
+        # type: (Optional[int], Any) -> ThroughputProperties
         """Replace the database-level throughput.
 
         :param throughput: The throughput to be set (an integer).
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the database, updated with new throughput.
+        :returns: ThroughputProperties for the database, updated with new throughput.
         :raises ~azure.cosmos.exceptions.CosmosHttpResponseError:
-            If no offer exists for the database or if the offer could not be updated.
-        :rtype: ~azure.cosmos.Offer
+            If no throughput properties exists for the database or if the throughput properties could not be updated.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = await self._get_properties()
@@ -733,15 +739,17 @@ async def replace_throughput(self, throughput, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = [offer async for offer in self.client_connection.QueryOffers(query_spec, **kwargs)]
-        if len(offers) == 0:
+        throughput_properties = [throughput async for throughput in
+                                 self.client_connection.QueryOffers(query_spec, **kwargs)]
+        if len(throughput_properties) == 0:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for database " + self.database_link)
+                message="Could not find ThroughputProperties for database " + self.database_link)
 
-        new_offer = offers[0].copy()
-        new_offer["content"]["offerThroughput"] = throughput
-        data = await self.client_connection.ReplaceOffer(offer_link=offers[0]["_self"], offer=offers[0], **kwargs)
+        new_throughput_properties = throughput_properties[0].copy()
+        new_throughput_properties["content"]["offerThroughput"] = throughput
+        data = await self.client_connection.ReplaceOffer(offer_link=throughput_properties[0]["_self"],
+                                                         offer=throughput_properties[0], **kwargs)
         if response_hook:
             response_hook(self.client_connection.last_response_headers, data)
-        return Offer(offer_throughput=data["content"]["offerThroughput"], properties=data)
+        return ThroughputProperties(offer_throughput=data["content"]["offerThroughput"], properties=data)
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py
index 1658f6e957d4..eb5ed589e42a 100644
--- a/sdk/cosmos/azure-cosmos/azure/cosmos/container.py
+++ b/sdk/cosmos/azure-cosmos/azure/cosmos/container.py
@@ -24,18 +24,20 @@
 
 from typing import Any, Dict, List, Optional, Union, Iterable, cast  # pylint: disable=unused-import
 
+import warnings
 from azure.core.tracing.decorator import distributed_trace  # type: ignore
 
 from ._cosmos_client_connection import CosmosClientConnection
 from ._base import build_options, validate_cache_staleness_value
 from .exceptions import CosmosResourceNotFoundError
 from .http_constants import StatusCodes
-from .offer import Offer
-from .scripts import ScriptsProxy
+from .offer import ThroughputProperties
 from .partition_key import NonePartitionKeyValue
+from .scripts import ScriptsProxy
 
 __all__ = ("ContainerProxy",)
 
+
 # pylint: disable=protected-access
 # pylint: disable=missing-client-constructor-parameter-credential,missing-client-constructor-parameter-kwargs
 
@@ -590,16 +592,35 @@ def delete_item(
 
     @distributed_trace
     def read_offer(self, **kwargs):
-        # type: (Any) -> Offer
-        """Read the Offer object for this container.
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this container.
+
+        If no ThroughputProperties already exist for the container, an exception is raised.
+
+        :keyword Callable response_hook: A callable invoked with the response metadata.
+        :returns: Throughput for the container.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
+        """
+        warnings.warn(
+            "read_offer is a deprecated method name, use get_throughput instead",
+            DeprecationWarning
+        )
+        return self.get_throughput(**kwargs)
+
+    @distributed_trace
+    def get_throughput(self, **kwargs):
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this container.
 
-        If no Offer already exists for the container, an exception is raised.
+        If no ThroughputProperties already exist for the container, an exception is raised.
 
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the container.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No offer exists for the container or
-            the offer could not be retrieved.
-        :rtype: ~azure.cosmos.Offer
+        :returns: Throughput for the container.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = self._get_properties()
@@ -608,30 +629,31 @@ def read_offer(self, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = list(self.client_connection.QueryOffers(query_spec, **kwargs))
-        if not offers:
+        throughput_properties = list(self.client_connection.QueryOffers(query_spec, **kwargs))
+        if not throughput_properties:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for container " + self.container_link)
+                message="Could not find ThroughputProperties for container " + self.container_link)
 
         if response_hook:
-            response_hook(self.client_connection.last_response_headers, offers)
+            response_hook(self.client_connection.last_response_headers, throughput_properties)
 
-        return Offer(offer_throughput=offers[0]["content"]["offerThroughput"], properties=offers[0])
+        return ThroughputProperties(offer_throughput=throughput_properties[0]["content"]["offerThroughput"],
+                                    properties=throughput_properties[0])
 
     @distributed_trace
     def replace_throughput(self, throughput, **kwargs):
-        # type: (int, Any) -> Offer
+        # type: (int, Any) -> ThroughputProperties
         """Replace the container's throughput.
 
-        If no Offer already exists for the container, an exception is raised.
+        If no ThroughputProperties already exist for the container, an exception is raised.
 
         :param throughput: The throughput to be set (an integer).
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the container, updated with new throughput.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No offer exists for the container
-            or the offer could not be updated.
-        :rtype: ~azure.cosmos.Offer
+        :returns: ThroughputProperties for the container, updated with new throughput.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exist for the container
+            or the throughput properties could not be updated.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = self._get_properties()
@@ -640,19 +662,20 @@ def replace_throughput(self, throughput, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = list(self.client_connection.QueryOffers(query_spec, **kwargs))
-        if not offers:
+        throughput_properties = list(self.client_connection.QueryOffers(query_spec, **kwargs))
+        if not throughput_properties:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for container " + self.container_link)
-        new_offer = offers[0].copy()
-        new_offer["content"]["offerThroughput"] = throughput
-        data = self.client_connection.ReplaceOffer(offer_link=offers[0]["_self"], offer=offers[0], **kwargs)
+                message="Could not find ThroughputProperties for container " + self.container_link)
+        new_throughput_properties = throughput_properties[0].copy()
+        new_throughput_properties["content"]["offerThroughput"] = throughput
+        data = self.client_connection.ReplaceOffer(
+            offer_link=throughput_properties[0]["_self"], offer=throughput_properties[0], **kwargs)
 
         if response_hook:
             response_hook(self.client_connection.last_response_headers, data)
 
-        return Offer(offer_throughput=data["content"]["offerThroughput"], properties=data)
+        return ThroughputProperties(offer_throughput=data["content"]["offerThroughput"], properties=data)
 
     @distributed_trace
     def list_conflicts(self, max_item_count=None, **kwargs):
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/database.py b/sdk/cosmos/azure-cosmos/azure/cosmos/database.py
index 566f00cfcb0c..36121d4f6635 100644
--- a/sdk/cosmos/azure-cosmos/azure/cosmos/database.py
+++ b/sdk/cosmos/azure-cosmos/azure/cosmos/database.py
@@ -30,7 +30,7 @@
 from ._cosmos_client_connection import CosmosClientConnection
 from ._base import build_options
 from .container import ContainerProxy
-from .offer import Offer
+from .offer import ThroughputProperties
 from .http_constants import StatusCodes
 from .exceptions import CosmosResourceNotFoundError
 from .user import UserProxy
@@ -708,14 +708,35 @@ def delete_user(self, user, **kwargs):
 
     @distributed_trace
     def read_offer(self, **kwargs):
-        # type: (Any) -> Offer
-        """Read the Offer object for this database.
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this database.
+
+        If no ThroughputProperties already exist for the database, an exception is raised.
 
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the database.
-        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError:
-            If no offer exists for the database or if the offer could not be retrieved.
-        :rtype: ~azure.cosmos.Offer
+        :returns: ThroughputProperties for the database.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
+        """
+        warnings.warn(
+            "read_offer is a deprecated method name, use read_throughput instead",
+            DeprecationWarning
+        )
+        return self.get_throughput(**kwargs)
+
+    @distributed_trace
+    def get_throughput(self, **kwargs):
+        # type: (Any) -> ThroughputProperties
+        """Get the ThroughputProperties object for this database.
+
+        If no ThroughputProperties already exist for the database, an exception is raised.
+
+        :keyword Callable response_hook: A callable invoked with the response metadata.
+        :returns: ThroughputProperties for the database.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: No throughput properties exists for the container or
+            the throughput properties could not be retrieved.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = self._get_properties()
@@ -724,28 +745,29 @@ def read_offer(self, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = list(self.client_connection.QueryOffers(query_spec, **kwargs))
-        if not offers:
+        throughput_properties = list(self.client_connection.QueryOffers(query_spec, **kwargs))
+        if not throughput_properties:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for database " + self.database_link)
+                message="Could not find ThroughputProperties for database " + self.database_link)
 
         if response_hook:
-            response_hook(self.client_connection.last_response_headers, offers)
+            response_hook(self.client_connection.last_response_headers, throughput_properties)
 
-        return Offer(offer_throughput=offers[0]["content"]["offerThroughput"], properties=offers[0])
+        return ThroughputProperties(offer_throughput=throughput_properties[0]["content"]["offerThroughput"],
+                                    properties=throughput_properties[0])
 
     @distributed_trace
     def replace_throughput(self, throughput, **kwargs):
-        # type: (Optional[int], Any) -> Offer
+        # type: (Optional[int], Any) -> ThroughputProperties
         """Replace the database-level throughput.
 
         :param throughput: The throughput to be set (an integer).
         :keyword Callable response_hook: A callable invoked with the response metadata.
-        :returns: Offer for the database, updated with new throughput.
+        :returns: ThroughputProperties for the database, updated with new throughput.
         :raises ~azure.cosmos.exceptions.CosmosHttpResponseError:
-            If no offer exists for the database or if the offer could not be updated.
-        :rtype: ~azure.cosmos.Offer
+            If no throughput properties exists for the database or if the throughput properties could not be updated.
+        :rtype: ~azure.cosmos.ThroughputProperties
         """
         response_hook = kwargs.pop('response_hook', None)
         properties = self._get_properties()
@@ -754,14 +776,15 @@ def replace_throughput(self, throughput, **kwargs):
             "query": "SELECT * FROM root r WHERE r.resource=@link",
             "parameters": [{"name": "@link", "value": link}],
         }
-        offers = list(self.client_connection.QueryOffers(query_spec))
-        if not offers:
+        throughput_properties = list(self.client_connection.QueryOffers(query_spec))
+        if not throughput_properties:
             raise CosmosResourceNotFoundError(
                 status_code=StatusCodes.NOT_FOUND,
-                message="Could not find Offer for collection " + self.database_link)
-        new_offer = offers[0].copy()
+                message="Could not find ThroughputProperties for database " + self.database_link)
+        new_offer = throughput_properties[0].copy()
         new_offer["content"]["offerThroughput"] = throughput
-        data = self.client_connection.ReplaceOffer(offer_link=offers[0]["_self"], offer=offers[0], **kwargs)
+        data = self.client_connection.ReplaceOffer(offer_link=throughput_properties[0]["_self"],
+                                                   offer=throughput_properties[0], **kwargs)
         if response_hook:
             response_hook(self.client_connection.last_response_headers, data)
-        return Offer(offer_throughput=data["content"]["offerThroughput"], properties=data)
+        return ThroughputProperties(offer_throughput=data["content"]["offerThroughput"], properties=data)
diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/offer.py b/sdk/cosmos/azure-cosmos/azure/cosmos/offer.py
index 4b99bf668055..e615bb94a7eb 100644
--- a/sdk/cosmos/azure-cosmos/azure/cosmos/offer.py
+++ b/sdk/cosmos/azure-cosmos/azure/cosmos/offer.py
@@ -1,5 +1,5 @@
 # The MIT License (MIT)
-# Copyright (c) 2014 Microsoft Corporation
+# Copyright (c) 2021 Microsoft Corporation
 
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
@@ -19,18 +19,21 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-"""Create offers in the Azure Cosmos DB SQL API service.
+"""Create throughput properties in the Azure Cosmos DB SQL API service.
 """
 from typing import Dict, Any
 
 
-class Offer(object):
-    """Represents a offer in an Azure Cosmos DB SQL API container.
+class ThroughputProperties(object):
+    """Represents the throughput properties in an Azure Cosmos DB SQL API container.
 
-    To read and update offers use the associated methods on the :class:`Container`.
+    To read and update throughput properties use the associated methods on the :class:`Container`.
     """
 
     def __init__(self, offer_throughput, properties=None):  # pylint: disable=super-init-not-called
         # type: (int, Dict[str, Any]) -> None
         self.offer_throughput = offer_throughput
         self.properties = properties
+
+
+Offer = ThroughputProperties
diff --git a/sdk/cosmos/azure-cosmos/samples/container_management.py b/sdk/cosmos/azure-cosmos/samples/container_management.py
index c826f9b4aa12..b3c55434eb48 100644
--- a/sdk/cosmos/azure-cosmos/samples/container_management.py
+++ b/sdk/cosmos/azure-cosmos/samples/container_management.py
@@ -196,7 +196,7 @@ def manage_provisioned_throughput(db, id):
         container = db.get_container_client(container=id)
 
         # now use its _self to query for Offers
-        offer = container.read_offer()
+        offer = container.get_throughput()
 
         print('Found Offer \'{0}\' for Container \'{1}\' and its throughput is \'{2}\''.format(offer.properties['id'], container.id, offer.properties['content']['offerThroughput']))
 
diff --git a/sdk/cosmos/azure-cosmos/samples/container_management_async.py b/sdk/cosmos/azure-cosmos/samples/container_management_async.py
index 4b3ab86a843d..959bb2ae805b 100644
--- a/sdk/cosmos/azure-cosmos/samples/container_management_async.py
+++ b/sdk/cosmos/azure-cosmos/samples/container_management_async.py
@@ -215,7 +215,7 @@ async def manage_provisioned_throughput(db, id):
         container = db.get_container_client(id)
 
         # now use its _self to query for throughput offers
-        offer = await container.read_offer()
+        offer = await container.get_throughput()
 
         print('Found Offer \'{0}\' for Container \'{1}\' and its throughput is \'{2}\''.format(offer.properties['id'], container.id, offer.properties['content']['offerThroughput']))
 
diff --git a/sdk/cosmos/azure-cosmos/test/test_backwards_compatibility.py b/sdk/cosmos/azure-cosmos/test/test_backwards_compatibility.py
new file mode 100644
index 000000000000..a4f345874bf0
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos/test/test_backwards_compatibility.py
@@ -0,0 +1,65 @@
+# The MIT License (MIT)
+# Copyright (c) 2022 Microsoft Corporation
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import unittest
+import pytest
+from azure.cosmos import cosmos_client, PartitionKey, Offer
+import test_config
+
+# This class tests the backwards compatibility of features being deprecated to ensure users are not broken before
+# properly removing the methods marked for deprecation.
+
+pytestmark = pytest.mark.cosmosEmulator
+
+
+@pytest.mark.usefixtures("teardown")
+class TestBackwardsCompatibility(unittest.TestCase):
+
+    configs = test_config._test_config
+    host = configs.host
+    masterKey = configs.masterKey
+
+    @classmethod
+    def setUpClass(cls):
+        if cls.masterKey == '[YOUR_KEY_HERE]' or cls.host == '[YOUR_ENDPOINT_HERE]':
+            raise Exception(
+                "You must specify your Azure Cosmos account values for "
+                "'masterKey' and 'host' at the top of this class to run the "
+                "tests.")
+        cls.client = cosmos_client.CosmosClient(cls.host, cls.masterKey, consistency_level="Session")
+        cls.databaseForTest = cls.client.create_database_if_not_exists(cls.configs.TEST_DATABASE_ID,
+                                                                       offer_throughput=500)
+        cls.containerForTest = cls.databaseForTest.create_container_if_not_exists(
+            cls.configs.TEST_COLLECTION_SINGLE_PARTITION_ID, PartitionKey(path="/id"), offer_throughput=400)
+
+    def test_offer_methods(self):
+        database_offer = self.databaseForTest.read_offer()
+        container_offer = self.containerForTest.read_offer()
+
+        self.assertTrue("ThroughputProperties" in str(type(database_offer)))
+        self.assertTrue("ThroughputProperties" in str(type(container_offer)))
+
+        self.assertTrue(isinstance(database_offer, Offer))
+        self.assertTrue(isinstance(container_offer, Offer))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/sdk/cosmos/azure-cosmos/test/test_crud.py b/sdk/cosmos/azure-cosmos/test/test_crud.py
index b0bedd3cc230..e2bdabba3d32 100644
--- a/sdk/cosmos/azure-cosmos/test/test_crud.py
+++ b/sdk/cosmos/azure-cosmos/test/test_crud.py
@@ -281,7 +281,7 @@ def test_partitioned_collection(self):
         self.assertEqual(collection_definition.get('partitionKey').get('kind'),
                          created_collection_properties['partitionKey']['kind'])
 
-        expected_offer = created_collection.read_offer()
+        expected_offer = created_collection.get_throughput()
 
         self.assertIsNotNone(expected_offer)
 
@@ -2283,14 +2283,14 @@ def test_offer_read_and_query(self):
             partition_key=PartitionKey(path='/id', kind='Hash')
         )
         # Read the offer.
-        expected_offer = collection.read_offer()
+        expected_offer = collection.get_throughput()
         collection_properties = collection.read()
         self.__ValidateOfferResponseBody(expected_offer, collection_properties.get('_self'), None)
 
         # Now delete the collection.
         db.delete_container(container=collection)
         # Reading fails.
-        self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, collection.read_offer)
+        self.__AssertHTTPFailureWithStatus(StatusCodes.NOT_FOUND, collection.get_throughput)
 
     def test_offer_replace(self):
         # Create database.
@@ -2298,7 +2298,7 @@ def test_offer_replace(self):
         # Create collection.
         collection = self.configs.create_multi_partition_collection_if_not_exist(self.client)
         # Read Offer
-        expected_offer = collection.read_offer()
+        expected_offer = collection.get_throughput()
         collection_properties = collection.read()
         self.__ValidateOfferResponseBody(expected_offer, collection_properties.get('_self'), None)
         # Replace the offer.
diff --git a/sdk/cosmos/azure-cosmos/test/test_partition_split_query.py b/sdk/cosmos/azure-cosmos/test/test_partition_split_query.py
index a00aadb36e4d..73beae54421a 100644
--- a/sdk/cosmos/azure-cosmos/test/test_partition_split_query.py
+++ b/sdk/cosmos/azure-cosmos/test/test_partition_split_query.py
@@ -60,11 +60,11 @@ def test_partition_split_query(self):
 
         self.run_queries(self.container, 500)  # initial check for queries before partition split
         print("initial check succeeded, now reading offer until replacing is done")
-        offer = self.database.read_offer()
+        offer = self.database.get_throughput()
         while True:
             if offer.properties['content'].get('isOfferReplacePending', False):
                 time.sleep(10)
-                offer = self.database.read_offer()
+                offer = self.database.get_throughput()
             else:
                 print("offer replaced successfully, took around {} seconds".format(time.time() - offer_time))
                 self.run_queries(self.container, 500)  # check queries work post partition split