diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 601125050..c354b4348 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -4171,6 +4171,16 @@
 - [X] put_object
 </details>
 
+## meteringmarketplace
+<details>
+<summary>25% implemented</summary>
+
+- [X] batch_meter_usage
+- [ ] meter_usage
+- [ ] register_usage
+- [ ] resolve_customer
+</details>
+
 ## mq
 <details>
 <summary>86% implemented</summary>
@@ -6480,7 +6490,6 @@
 - mediapackage-vod
 - mediatailor
 - memorydb
-- meteringmarketplace
 - mgh
 - mgn
 - migration-hub-refactor-spaces
diff --git a/docs/docs/services/meteringmarketplace.rst b/docs/docs/services/meteringmarketplace.rst
new file mode 100644
index 000000000..36d0cfe86
--- /dev/null
+++ b/docs/docs/services/meteringmarketplace.rst
@@ -0,0 +1,32 @@
+.. _implementedservice_meteringmarketplace:
+
+.. |start-h3| raw:: html
+
+    <h3>
+
+.. |end-h3| raw:: html
+
+    </h3>
+
+===================
+meteringmarketplace
+===================
+
+|start-h3| Example usage |end-h3|
+
+.. sourcecode:: python
+
+            @mock_meteringmarketplace
+            def test_meteringmarketplace_behaviour:
+                boto3.client("meteringmarketplace")
+                ...
+
+
+
+|start-h3| Implemented features for this service |end-h3|
+
+- [X] batch_meter_usage
+- [ ] meter_usage
+- [ ] register_usage
+- [ ] resolve_customer
+
diff --git a/moto/__init__.py b/moto/__init__.py
index 2f594f16f..5ecd606f8 100644
--- a/moto/__init__.py
+++ b/moto/__init__.py
@@ -106,6 +106,7 @@ mock_mediastore = lazy_load(".mediastore", "mock_mediastore")
 mock_mediastoredata = lazy_load(
     ".mediastoredata", "mock_mediastoredata", boto3_name="mediastore-data"
 )
+mock_meteringmarketplace = lazy_load(".meteringmarketplace", "mock_meteringmarketplace")
 mock_mq = lazy_load(".mq", "mock_mq", boto3_name="mq")
 mock_opsworks = lazy_load(".opsworks", "mock_opsworks")
 mock_organizations = lazy_load(".organizations", "mock_organizations")
diff --git a/moto/backend_index.py b/moto/backend_index.py
index 0e16a849d..d634817fd 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -102,6 +102,11 @@ backend_url_patterns = [
         "mediastore-data",
         re.compile("https?://data\\.mediastore\\.(.+)\\.amazonaws.com"),
     ),
+    (
+        "meteringmarketplace",
+        re.compile("https?://metering.marketplace.(.+).amazonaws.com"),
+    ),
+    ("meteringmarketplace", re.compile("https?://aws-marketplace.(.+).amazonaws.com")),
     ("mq", re.compile("https?://mq\\.(.+)\\.amazonaws\\.com")),
     ("opsworks", re.compile("https?://opsworks\\.us-east-1\\.amazonaws.com")),
     ("organizations", re.compile("https?://organizations\\.(.+)\\.amazonaws\\.com")),
diff --git a/moto/meteringmarketplace/__init__.py b/moto/meteringmarketplace/__init__.py
new file mode 100644
index 000000000..cb50b9840
--- /dev/null
+++ b/moto/meteringmarketplace/__init__.py
@@ -0,0 +1,6 @@
+from __future__ import unicode_literals
+from .models import meteringmarketplace_backends
+from ..core.models import base_decorator
+
+meteringmarketplace_backend = meteringmarketplace_backends["us-east-1"]
+mock_meteringmarketplace = base_decorator(meteringmarketplace_backends)
diff --git a/moto/meteringmarketplace/exceptions.py b/moto/meteringmarketplace/exceptions.py
new file mode 100644
index 000000000..25fcc98e6
--- /dev/null
+++ b/moto/meteringmarketplace/exceptions.py
@@ -0,0 +1,60 @@
+from __future__ import unicode_literals
+
+import json
+from werkzeug.exceptions import BadRequest
+
+
+class DisabledApiException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "DisabledApiException"}
+        )
+
+
+class InternalServiceErrorException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "InternalServiceErrorException"}
+        )
+
+
+class InvalidCustomerIdentifierException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "InvalidCustomerIdentifierException"}
+        )
+
+
+class InvalidProductCodeException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "InvalidProductCodeException"}
+        )
+
+
+class InvalidUsageDimensionException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "InvalidUsageDimensionException"}
+        )
+
+
+class ThrottlingException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "ThrottlingException"}
+        )
+
+
+class TimestampOutOfBoundsException(BadRequest):
+    def __init__(self, message):
+        super().__init__()
+        self.description = json.dumps(
+            {"message": message, "__type": "TimestampOutOfBoundsException"}
+        )
diff --git a/moto/meteringmarketplace/models.py b/moto/meteringmarketplace/models.py
new file mode 100644
index 000000000..b08bfcfa7
--- /dev/null
+++ b/moto/meteringmarketplace/models.py
@@ -0,0 +1,143 @@
+import collections
+from moto.core import BaseBackend, BaseModel
+from moto.moto_api._internal import mock_random
+from moto.core.utils import BackendDict
+
+
+class UsageRecord(BaseModel, dict):
+    def __init__(self, timestamp, customer_identifier, dimension, quantity=0):
+        super().__init__()
+        self.timestamp = timestamp
+        self.customer_identifier = customer_identifier
+        self.dimension = dimension
+        self.quantity = quantity
+        self.metering_record_id = mock_random.uuid4().hex
+
+    @classmethod
+    def from_data(cls, data):
+        cls(
+            timestamp=data.get("Timestamp"),
+            customer_identifier=data.get("CustomerIdentifier"),
+            dimension=data.get("Dimension"),
+            quantity=data.get("Quantity", 0),
+        )
+
+    @property
+    def timestamp(self):
+        return self["Timestamp"]
+
+    @timestamp.setter
+    def timestamp(self, value):
+        self["Timestamp"] = value
+
+    @property
+    def customer_identifier(self):
+        return self["CustomerIdentifier"]
+
+    @customer_identifier.setter
+    def customer_identifier(self, value):
+        self["CustomerIdentifier"] = value
+
+    @property
+    def dimension(self):
+        return self["Dimension"]
+
+    @dimension.setter
+    def dimension(self, value):
+        self["Dimension"] = value
+
+    @property
+    def quantity(self):
+        return self["Quantity"]
+
+    @quantity.setter
+    def quantity(self, value):
+        self["Quantity"] = value
+
+
+class Result(BaseModel, dict):
+    SUCCESS = "Success"
+    CUSTOMER_NOT_SUBSCRIBED = "CustomerNotSubscribed"
+    DUPLICATE_RECORD = "DuplicateRecord"
+
+    def __init__(self, **kwargs):
+        self.usage_record = UsageRecord(
+            timestamp=kwargs["Timestamp"],
+            customer_identifier=kwargs["CustomerIdentifier"],
+            dimension=kwargs["Dimension"],
+            quantity=kwargs["Quantity"],
+        )
+        self.status = Result.SUCCESS
+        self["MeteringRecordId"] = self.usage_record.metering_record_id
+
+    @property
+    def metering_record_id(self):
+        return self["MeteringRecordId"]
+
+    @property
+    def status(self):
+        return self["Status"]
+
+    @status.setter
+    def status(self, value):
+        self["Status"] = value
+
+    @property
+    def usage_record(self):
+        return self["UsageRecord"]
+
+    @usage_record.setter
+    def usage_record(self, value):
+        if not isinstance(value, UsageRecord):
+            value = UsageRecord.from_data(value)
+        self["UsageRecord"] = value
+
+    def is_duplicate(self, other):
+        """
+        DuplicateRecord - Indicates that the UsageRecord was invalid and not honored.
+        A previously metered UsageRecord had the same customer, dimension, and time,
+        but a different quantity.
+        """
+        assert isinstance(other, Result), "Needs to be a Result type"
+        usage_record, other = other.usage_record, self.usage_record
+        return (
+            other.customer_identifier == usage_record.customer_identifier
+            and other.dimension == usage_record.dimension
+            and other.timestamp == usage_record.timestamp
+            and other.quantity != usage_record.quantity
+        )
+
+
+class CustomerDeque(collections.deque):
+    def is_subscribed(self, customer):
+        return customer in self
+
+
+class ResultDeque(collections.deque):
+    def is_duplicate(self, result):
+        return any(record.is_duplicate(result) for record in self)
+
+
+class MeteringMarketplaceBackend(BaseBackend):
+    def __init__(self, region_name, account_id):
+        super().__init__(region_name, account_id)
+        self.customers_by_product = collections.defaultdict(CustomerDeque)
+        self.records_by_product = collections.defaultdict(ResultDeque)
+
+    def batch_meter_usage(self, product_code, usage_records):
+        results = []
+        for usage in usage_records:
+            result = Result(**usage)
+            if not self.customers_by_product[product_code].is_subscribed(
+                result.usage_record.customer_identifier
+            ):
+                result.status = result.CUSTOMER_NOT_SUBSCRIBED
+            elif self.records_by_product[product_code].is_duplicate(result):
+                result.status = result.DUPLICATE_RECORD
+            results.append(result)
+        return results
+
+
+meteringmarketplace_backends = BackendDict(
+    MeteringMarketplaceBackend, "meteringmarketplace"
+)
diff --git a/moto/meteringmarketplace/responses.py b/moto/meteringmarketplace/responses.py
new file mode 100644
index 000000000..d3addb966
--- /dev/null
+++ b/moto/meteringmarketplace/responses.py
@@ -0,0 +1,17 @@
+import json
+
+from moto.core.responses import BaseResponse
+from .models import meteringmarketplace_backends, MeteringMarketplaceBackend
+
+
+class MarketplaceMeteringResponse(BaseResponse):
+    @property
+    def backend(self) -> MeteringMarketplaceBackend:
+        return meteringmarketplace_backends[self.current_account][self.region]
+
+    def batch_meter_usage(self):
+        results = []
+        usage_records = json.loads(self.body)["UsageRecords"]
+        product_code = json.loads(self.body)["ProductCode"]
+        results = self.backend.batch_meter_usage(product_code, usage_records)
+        return json.dumps({"Results": results, "UnprocessedRecords": []})
diff --git a/moto/meteringmarketplace/urls.py b/moto/meteringmarketplace/urls.py
new file mode 100644
index 000000000..a351525e2
--- /dev/null
+++ b/moto/meteringmarketplace/urls.py
@@ -0,0 +1,9 @@
+from __future__ import unicode_literals
+from .responses import MarketplaceMeteringResponse
+
+url_bases = [
+    "https?://metering.marketplace.(.+).amazonaws.com",
+    "https?://aws-marketplace.(.+).amazonaws.com",
+]
+
+url_paths = {"{0}/$": MarketplaceMeteringResponse.dispatch}
diff --git a/tests/test_meteringmarketplace/test_meteringmarketplace.py b/tests/test_meteringmarketplace/test_meteringmarketplace.py
new file mode 100644
index 000000000..cbc5928a9
--- /dev/null
+++ b/tests/test_meteringmarketplace/test_meteringmarketplace.py
@@ -0,0 +1,95 @@
+import boto3
+import copy
+import sure  # noqa # pylint: disable=unused-import
+from datetime import datetime
+from moto import mock_meteringmarketplace
+from moto.meteringmarketplace.models import Result
+
+
+USAGE_RECORDS = [
+    {
+        "Timestamp": datetime(2019, 8, 25, 21, 1, 38),
+        "CustomerIdentifier": "EqCDbVpkMBaQzPQv",
+        "Dimension": "HIDaByYzIxgGZqyz",
+        "Quantity": 6984,
+    },
+    {
+        "Timestamp": datetime(2019, 9, 7, 16, 4, 47),
+        "CustomerIdentifier": "ITrDnpWEiebXybRJ",
+        "Dimension": "BSWatHqyhbyTQHCS",
+        "Quantity": 6388,
+    },
+    {
+        "Timestamp": datetime(2019, 6, 15, 23, 17, 49),
+        "CustomerIdentifier": "YfzTVRheDsXEehgQ",
+        "Dimension": "FsVxwLkTAynaWwGT",
+        "Quantity": 3532,
+    },
+    {
+        "Timestamp": datetime(2019, 9, 10, 19, 56, 35),
+        "CustomerIdentifier": "kmgmFPcWhpDGSSDm",
+        "Dimension": "PBXGdBHQWVOwudRK",
+        "Quantity": 9897,
+    },
+    {
+        "Timestamp": datetime(2019, 1, 12, 1, 28, 36),
+        "CustomerIdentifier": "OyxECDjaaDVUeQIB",
+        "Dimension": "VzwLdmFjbTBBbHJg",
+        "Quantity": 5142,
+    },
+    {
+        "Timestamp": datetime(2019, 8, 5, 18, 27, 41),
+        "CustomerIdentifier": "PkeNkaJVfceGvYAX",
+        "Dimension": "mHTtIbsLAYrCVSNM",
+        "Quantity": 6503,
+    },
+    {
+        "Timestamp": datetime(2019, 7, 18, 3, 22, 18),
+        "CustomerIdentifier": "ARzQRYYuXGHCVDkW",
+        "Dimension": "RMjZbVpxehPTNtDL",
+        "Quantity": 5465,
+    },
+    {
+        "Timestamp": datetime(2019, 6, 24, 9, 19, 14),
+        "CustomerIdentifier": "kzlvIayJzpyfBPaH",
+        "Dimension": "MmsydkBEHERXDooi",
+        "Quantity": 6135,
+    },
+    {
+        "Timestamp": datetime(2019, 9, 28, 20, 29, 5),
+        "CustomerIdentifier": "nygdXONtkiqktTMn",
+        "Dimension": "GKbTFPMGRyZObEEz",
+        "Quantity": 3416,
+    },
+    {
+        "Timestamp": datetime(2019, 6, 17, 2, 5, 34),
+        "CustomerIdentifier": "JlIGIXHHPphnBhfV",
+        "Dimension": "dcBPhccyHYSdPUUO",
+        "Quantity": 2184,
+    },
+]
+
+
+@mock_meteringmarketplace()
+def test_batch_meter_usage():
+    client = boto3.client("meteringmarketplace", region_name="us-east-1")
+
+    res = client.batch_meter_usage(
+        UsageRecords=USAGE_RECORDS, ProductCode="PUFXZLyUElvQvrsG"
+    )
+
+    res.should.have.key("Results").length_of(10)
+
+    records_without_time = copy.copy(USAGE_RECORDS)
+    for r in records_without_time:
+        r.pop("Timestamp")
+
+    for record in res["Results"]:
+        record["UsageRecord"].pop("Timestamp")
+        assert record["UsageRecord"] in records_without_time
+        assert record["MeteringRecordId"]
+        assert record["Status"] in [
+            Result.DUPLICATE_RECORD,
+            Result.CUSTOMER_NOT_SUBSCRIBED,
+            Result.SUCCESS,
+        ]