diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index cd5cc357..efcaa7d5 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -4,6 +4,7 @@ CHANGELOG
 
 Unreleased
 ==========
+* bugfix: Record these values in exception messages as `REDACTED`: the values of sensitive url query parameters, e.g. of presigned S3 urls, and, unless a pymongo listener records full documents (`record_full_documents=True`), the values of pymongo error documents other than `ok`, `code`, `codeName`, `errorLabels`, `errtype` and the position of write errors in a bulk write, as in the failure metadata, and MongoDB duplicate key values.
 * bugfix: [pymongo] Only record `ok`, `code`, `codeName`, `errorLabels` and `errtype` of failed commands unless `record_full_documents=True`. `errmsg` and other failure fields can contain document values and are no longer recorded by default. `https://github.com/aws/aws-xray-sdk-python/pull/495`
 
 2.15.0
diff --git a/aws_xray_sdk/core/models/throwable.py b/aws_xray_sdk/core/models/throwable.py
index 21d6483f..6dc21a3b 100644
--- a/aws_xray_sdk/core/models/throwable.py
+++ b/aws_xray_sdk/core/models/throwable.py
@@ -2,10 +2,92 @@
 import os
 import binascii
 import logging
+from collections.abc import Mapping
+
+from ..utils.redaction import REDACTED, mask
 
 log = logging.getLogger(__name__)
 
 
+# fields of pymongo error documents that never contain document values, as in the failure metadata
+_SAFE_FAILURE_FIELDS = ('ok', 'code', 'codeName', 'errorLabels', 'errtype')
+
+# and the position of a write error in a bulk write
+_SAFE_WRITE_ERROR_FIELDS = _SAFE_FAILURE_FIELDS + ('index', 'idx')
+
+# errors in an error document, and their fields that never contain document values
+_ERRORS = {
+    'writeErrors': _SAFE_WRITE_ERROR_FIELDS,
+    'writeConcernErrors': _SAFE_FAILURE_FIELDS,
+    'writeConcernError': _SAFE_FAILURE_FIELDS,
+}
+
+
+def _redact_error_document(document, safe_fields=_SAFE_FAILURE_FIELDS):
+    redacted = {}
+    for key, value in document.items():
+        if key in safe_fields:
+            redacted[key] = value
+        elif key in _ERRORS:
+            redacted[key] = _redact_errors(value, _ERRORS[key])
+        else:
+            redacted[key] = REDACTED
+    return redacted
+
+
+def _redact_errors(value, safe_fields):
+    if isinstance(value, Mapping):
+        return _redact_error_document(value, safe_fields)
+    if isinstance(value, (list, tuple)):
+        return [_redact_error_document(error, safe_fields) if isinstance(error, Mapping) else REDACTED
+                for error in value]
+    return REDACTED
+
+
+def _redact_dup_key(message):
+    # duplicate key error messages end with the key values, e.g. 'dup key: { email: "..." }'
+    i = message.find('dup key: ')
+    return message if i < 0 else message[:i] + 'dup key: ' + REDACTED
+
+
+def _redact_pymongo_error_document(exception, message):
+    if not any(str(getattr(cls, '__module__', '')).startswith('pymongo.') for cls in type(exception).__mro__):
+        return message
+    # pymongo errors include the server error document, and errors such as EncryptionError the document of the
+    # error that caused them
+    head, separator, _ = message.partition(', full error: ')
+    details = getattr(exception, 'details', None)
+    for name in ('__cause__', 'cause', '__context__'):
+        if isinstance(details, Mapping):
+            break
+        details = getattr(getattr(exception, name, None), 'details', None)
+    if not separator or not isinstance(details, Mapping):
+        return _redact_dup_key(message)
+    # the server message is kept, except duplicate key values
+    return _redact_dup_key(head) + separator + str(_redact_error_document(details))
+
+
+def _mask_message(exception, message):
+    return mask(message)
+
+
+# rules that redact sensitive data in exception messages, each a function of the exception and its message
+_REDACTION_RULES = (_redact_pymongo_error_document, _mask_message)
+
+
+def _record_full_pymongo_documents():
+    # pymongo exception messages are recorded in full, in the whole process, once a pymongo listener records
+    # full documents, see aws_xray_sdk.ext.pymongo.patch.XrayCommandListener
+    global _REDACTION_RULES
+    _REDACTION_RULES = tuple(rule for rule in _REDACTION_RULES if rule is not _redact_pymongo_error_document)
+
+
+def _redact(exception, message):
+    for rule in _REDACTION_RULES:
+        message = rule(exception, message)
+    return message
+
+
 class Throwable:
     """
     An object recording exception infomation under trace entity
@@ -23,8 +105,8 @@ def __init__(self, exception, stack, remote=False):
         self.id = binascii.b2a_hex(os.urandom(8)).decode('utf-8')
 
         try:
-            message = str(exception)
-            # in case there is an exception cannot be converted to str
+            message = _redact(exception, str(exception))
+            # in case there is an exception cannot be converted to str or redacted
         except Exception:
             message = None
 
diff --git a/aws_xray_sdk/core/utils/redaction.py b/aws_xray_sdk/core/utils/redaction.py
new file mode 100644
index 00000000..1f9b375e
--- /dev/null
+++ b/aws_xray_sdk/core/utils/redaction.py
@@ -0,0 +1,27 @@
+import re
+
+# the placeholder of masked data
+REDACTED = 'REDACTED'
+
+# url query parameters with sensitive values, e.g. of presigned urls
+_SENSITIVE_QUERY_PARAMETERS = (
+    'AWSAccessKeyId', 'Signature', 'sig', 'X-Amz-Credential', 'X-Amz-Security-Token', 'X-Amz-Signature',
+    'X-Goog-Signature', 'token',
+)
+
+# patterns of sensitive data that is masked in exception messages, each a text that messages matching the pattern
+# contain, to skip other messages quickly, a regular expression and its replacement. Messages can include data sent
+# by clients, so patterns must not backtrack.
+_PATTERNS = (
+    # values of sensitive url query parameters, keeping the keys, e.g. 'X-Amz-Signature=REDACTED'
+    ('=', re.compile(r'([?&](?:%s)=)[^&\s#]*' % '|'.join(map(re.escape, _SENSITIVE_QUERY_PARAMETERS)), re.IGNORECASE),
+     r'\g<1>' + REDACTED),
+)
+
+
+def mask(text):
+    """Return the text with the data that matches the patterns masked."""
+    for hint, pattern, replacement in _PATTERNS:
+        if hint in text:
+            text = pattern.sub(replacement, text)
+    return text
diff --git a/aws_xray_sdk/ext/pymongo/patch.py b/aws_xray_sdk/ext/pymongo/patch.py
index be520cbf..867f9b11 100644
--- a/aws_xray_sdk/ext/pymongo/patch.py
+++ b/aws_xray_sdk/ext/pymongo/patch.py
@@ -3,9 +3,7 @@
 
 from pymongo import monitoring
 from aws_xray_sdk.core import xray_recorder
-
-# failure fields that never contain document values
-_SAFE_FAILURE_FIELDS = ('ok', 'code', 'codeName', 'errorLabels', 'errtype')
+from aws_xray_sdk.core.models.throwable import _SAFE_FAILURE_FIELDS, _record_full_pymongo_documents
 
 
 class XrayCommandListener(monitoring.CommandListener):
@@ -17,12 +15,17 @@ class XrayCommandListener(monitoring.CommandListener):
     records all available information provided by pymongo,
     except for `command`, `reply` and failure details.
     They may contain business secrets.
-    If you insist to record them, specify `record_full_documents=True`.
+    If you insist to record them, specify `record_full_documents=True`,
+    which also stops the redaction of pymongo exception messages, i.e. of
+    error documents and duplicate key values, in the whole process, from
+    when the listener is created.
     """
 
     def __init__(self, record_full_documents):
         super().__init__()
         self.record_full_documents = record_full_documents
+        if record_full_documents:
+            _record_full_pymongo_documents()
 
     def started(self, event):
         host, port = event.connection_id
diff --git a/docs/thirdparty.rst b/docs/thirdparty.rst
index 7bfe2acd..2bf369d3 100644
--- a/docs/thirdparty.rst
+++ b/docs/thirdparty.rst
@@ -100,4 +100,23 @@ patch directly before patching other modules::
 
     patch(record_full_documents=True)
 
-Note that this records every command, reply and failure document.
+Note that this records every command, reply and failure document. A listener created with
+``record_full_documents=True`` also stops the redaction of pymongo exception messages, that is of
+error documents and duplicate key values, see `Exception messages`_, in the whole process, from when
+it is created. Url query parameters are still redacted.
+
+Exception messages
+------------------
+
+The SDK records these values in exception messages as ``REDACTED``:
+
+* the values of sensitive url query parameters, keeping the keys, for example the signature of a
+  presigned S3 url, e.g. ``/object.txt?X-Amz-Signature=REDACTED``: ``AWSAccessKeyId``, ``Signature``,
+  ``sig``, ``X-Amz-Credential``, ``X-Amz-Security-Token``, ``X-Amz-Signature``, ``X-Goog-Signature``
+  and ``token``. A value ends at the next space, ``&`` or ``#``.
+* in pymongo errors, the values of the server error document, for example the value that failed
+  document validation (``consideredValue``), except the fields recorded for failed commands and the
+  position of write errors in a bulk write, also when another pymongo error includes the document, and
+  MongoDB duplicate key values, that is the rest of the server message after ``dup key:``. The server
+  message before the document is otherwise kept, and can include other values. Once a pymongo listener
+  records full documents, see above, these values are not redacted.
diff --git a/tests/ext/pymongo/test_pymongo.py b/tests/ext/pymongo/test_pymongo.py
index 6abaf485..3abda83f 100644
--- a/tests/ext/pymongo/test_pymongo.py
+++ b/tests/ext/pymongo/test_pymongo.py
@@ -4,9 +4,11 @@
 from bson import encode
 from bson.raw_bson import RawBSONDocument
 from pymongo import monitoring
+from pymongo.errors import BulkWriteError, DuplicateKeyError, OperationFailure, WriteError
 
 from aws_xray_sdk.core import xray_recorder
 from aws_xray_sdk.core.context import Context
+from aws_xray_sdk.core.models import throwable
 from aws_xray_sdk.ext.pymongo.patch import XrayCommandListener, patch
 
 SECRET = 'stored-secret-value'
@@ -70,6 +72,12 @@
 }
 
 
+@pytest.fixture(autouse=True)
+def redact_exception_messages(monkeypatch):
+    # listeners that record full documents record exception messages in full in the whole process
+    monkeypatch.setattr(throwable, '_REDACTION_RULES', throwable._REDACTION_RULES)
+
+
 @pytest.fixture(autouse=True)
 def construct_ctx():
     """
@@ -190,3 +198,34 @@ def test_patch_does_not_record_full_documents_by_default(monkeypatch):
         vars(monitoring).pop('_xray_enabled', None)
 
     assert [listener.record_full_documents for listener in registered] == [False]
+
+
+@pytest.mark.parametrize('new_error', [
+    lambda: OperationFailure('Document failed validation', 121, VALIDATION_FAILURE),
+    lambda: DuplicateKeyError(DUPLICATE_KEY_FAILURE['errmsg'], 11000, DUPLICATE_KEY_FAILURE),
+    # insert and update commands report validation failures as write errors
+    lambda: WriteError('Document failed validation', 121,
+                       {'index': 0, 'code': 121, 'errInfo': VALIDATION_FAILURE['errInfo']}),
+    lambda: BulkWriteError({'writeErrors': [{'index': 0, 'code': 121, 'errInfo': VALIDATION_FAILURE['errInfo'],
+                                             'op': {'token': SECRET}}]}),
+], ids=['OperationFailure', 'DuplicateKeyError', 'WriteError', 'BulkWriteError'])
+def test_exception_message_does_not_record_document_values(new_error):
+    error = new_error()
+    with pytest.raises(type(error)):
+        with xray_recorder.in_subsegment('app'):
+            raise error
+    subsegment = xray_recorder.current_segment().subsegments[0]
+
+    message = subsegment.cause['exceptions'][0].message
+    assert ", full error: {" in message and "'REDACTED'" in message
+    assert SECRET not in subsegment.serialize()
+
+
+def test_exception_message_records_document_values_with_full_documents():
+    XrayCommandListener(record_full_documents=True)
+    with pytest.raises(OperationFailure):
+        with xray_recorder.in_subsegment('app'):
+            raise OperationFailure('Document failed validation', 121, VALIDATION_FAILURE)
+    subsegment = xray_recorder.current_segment().subsegments[0]
+
+    assert SECRET in subsegment.cause['exceptions'][0].message
diff --git a/tests/ext/requests/test_requests.py b/tests/ext/requests/test_requests.py
index a31e108e..92a04d09 100644
--- a/tests/ext/requests/test_requests.py
+++ b/tests/ext/requests/test_requests.py
@@ -1,3 +1,5 @@
+import json
+
 import pytest
 import requests
 
@@ -157,3 +159,15 @@ def test_strip_http_url():
     assert http_meta['request']['method'].upper() == 'GET'
     assert http_meta['response']['status'] == status_code
 
+
+def test_exception_message_does_not_record_query_values(monkeypatch):
+    # the message has the path of the url when there is no proxy
+    monkeypatch.setenv('NO_PROXY', '*')
+    monkeypatch.setenv('no_proxy', '*')
+    with pytest.raises(requests.exceptions.ConnectionError):
+        requests.get('http://127.0.0.1:1/private/object.txt?X-Amz-Signature=signature&token=secret', timeout=1)
+    subsegment = xray_recorder.current_segment().subsegments[0]
+
+    message = json.loads(subsegment.serialize())['cause']['exceptions'][0]['message']
+    assert 'with url: /private/object.txt?X-Amz-Signature=REDACTED&token=REDACTED (Caused by' in message
+
diff --git a/tests/test_redaction.py b/tests/test_redaction.py
new file mode 100644
index 00000000..796fd11b
--- /dev/null
+++ b/tests/test_redaction.py
@@ -0,0 +1,44 @@
+import time
+
+import pytest
+
+from aws_xray_sdk.core.utils.redaction import mask
+
+SECRET = 'stored-secret-value'
+
+
+@pytest.mark.parametrize('text, masked', [
+    # a presigned url in a requests error
+    ("HTTPSConnectionPool(host='bucket.s3.amazonaws.com', port=443): Max retries exceeded with url: "
+     "/private/object.txt?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIAIOSFODNN7EXAMPLE%2F20261020"
+     "&X-Amz-Security-Token={0}&X-Amz-Signature={0} (Caused by ConnectTimeoutError())".format(SECRET),
+     "HTTPSConnectionPool(host='bucket.s3.amazonaws.com', port=443): Max retries exceeded with url: "
+     "/private/object.txt?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=REDACTED"
+     "&X-Amz-Security-Token=REDACTED&X-Amz-Signature=REDACTED (Caused by ConnectTimeoutError())"),
+    # Signature Version 2
+    ('/key?AWSAccessKeyId=AKIAIOSFODNN7EXAMPLE&Expires=1&Signature=%s&x-amz-security-token=%s' % (SECRET, SECRET),
+     '/key?AWSAccessKeyId=REDACTED&Expires=1&Signature=REDACTED&x-amz-security-token=REDACTED'),
+    ("Client error '404 Not Found' for url 'https://host/key?page=2&token=%s#top'" % SECRET,
+     "Client error '404 Not Found' for url 'https://host/key?page=2&token=REDACTED#top'"),
+    # values end at a space, '&' or '#', so the quote after a url is masked too
+    ("404, message='Not Found', url='https://host/key?token=public'%s'" % SECRET,
+     "404, message='Not Found', url='https://host/key?token=REDACTED"),
+    # other parameters and text are recorded as is
+    ('https://host/key?access_token=a&page=2', 'https://host/key?access_token=a&page=2'),
+    ('Did you mean this? yes', 'Did you mean this? yes'),
+])
+def test_sensitive_query_values(text, masked):
+    assert mask(text) == masked
+
+
+@pytest.mark.parametrize('text', [
+    '?token=' * 20000,
+    '&X-Amz-Signature' * 10000 + '=',
+    '?' + 'a' * 80000 + '=',
+], ids=['values', 'keys', 'long key'])
+def test_masking_time_is_linear(text):
+    # messages can include data sent by clients
+    start = time.perf_counter()
+    mask(text)
+
+    assert time.perf_counter() - start < 0.5
diff --git a/tests/test_throwable.py b/tests/test_throwable.py
index 2efe7978..3fa4af60 100644
--- a/tests/test_throwable.py
+++ b/tests/test_throwable.py
@@ -1,3 +1,6 @@
+import pytest
+
+from aws_xray_sdk.core.models import throwable
 from aws_xray_sdk.core.models.throwable import Throwable
 
 
@@ -28,3 +31,159 @@ def test_stack_trace_parsing():
     assert entry2['path'] == 'test.py'
     assert entry2['line'] == 3
     assert entry2['label'] == 'another_function'
+
+
+SECRET = 'stored-secret-value'
+
+
+class _OperationFailure(Exception):
+    """A pymongo error with the server error document"""
+    __module__ = 'pymongo.errors'
+
+    def __init__(self, message, details):
+        super().__init__(message if details is None else '{}, full error: {}'.format(message, details))
+        self.details = details
+
+
+def test_pymongo_error_document_values():
+    errmsg = 'E11000 duplicate key error collection: db.users index: email_1 dup key: { email: "%s" }' % SECRET
+    error = _OperationFailure(errmsg, {
+        'ok': 0.0, 'errmsg': errmsg, 'code': 11000, 'codeName': 'DuplicateKey', 'errorLabels': ['label'],
+        'keyPattern': {'email': 1}, 'keyValue': {'email': SECRET}, 'idx': SECRET,
+        'writeErrors': [{'index': 0, 'code': 11000, 'op': {'email': SECRET}}, SECRET],
+        'writeConcernError': {'code': 64, 'codeName': 'WriteConcernFailed', 'errInfo': {'wtimeout': True}},
+        'writeConcernErrors': [{'code': 64, 'index': SECRET}],
+    })
+
+    # like the failure metadata, only fields that never contain document values are kept
+    assert Throwable(error, None).message == (
+        "E11000 duplicate key error collection: db.users index: email_1 dup key: REDACTED, full error: "
+        "{'ok': 0.0, 'errmsg': 'REDACTED', 'code': 11000, 'codeName': 'DuplicateKey', 'errorLabels': ['label'], "
+        "'keyPattern': 'REDACTED', 'keyValue': 'REDACTED', 'idx': 'REDACTED', "
+        "'writeErrors': [{'index': 0, 'code': 11000, 'op': 'REDACTED'}, 'REDACTED'], "
+        "'writeConcernError': {'code': 64, 'codeName': 'WriteConcernFailed', 'errInfo': 'REDACTED'}, "
+        "'writeConcernErrors': [{'code': 64, 'index': 'REDACTED'}]}")
+
+
+def test_pymongo_validation_error_values():
+    error = _OperationFailure('Document failed validation', {
+        'ok': 0.0, 'errmsg': 'Document failed validation', 'code': 121, 'codeName': 'DocumentValidationFailure',
+        'errInfo': {'details': {'operatorName': '$jsonSchema', 'consideredValue': SECRET}},
+    })
+
+    assert Throwable(error, None).message == (
+        "Document failed validation, full error: {'ok': 0.0, 'errmsg': 'REDACTED', 'code': 121, "
+        "'codeName': 'DocumentValidationFailure', 'errInfo': 'REDACTED'}")
+
+
+def test_pymongo_errors_caused_by_other_errors():
+    class EncryptionError(Exception):
+        """A pymongo error with the message of the error that caused it"""
+        __module__ = 'pymongo.errors'
+
+    try:
+        try:
+            raise _OperationFailure('Document failed validation', {'code': 121, 'errInfo': {'consideredValue': SECRET}})
+        except _OperationFailure as e:
+            raise EncryptionError(str(e)) from e
+    except EncryptionError as error:
+        assert Throwable(error, None).message == (
+            "Document failed validation, full error: {'code': 121, 'errInfo': 'REDACTED'}")
+
+
+def test_pymongo_errors_with_the_document_of_another_error():
+    class NetworkTimeout(Exception):
+        """A pymongo error with the message of another error, and errors as details"""
+        __module__ = 'pymongo.errors'
+        details = []
+
+    class EncryptionError(Exception):
+        """A pymongo error with the message of the error that caused it"""
+        __module__ = 'pymongo.errors'
+
+        def __init__(self, cause):
+            super().__init__(str(cause))
+            self.cause = cause
+
+    cause = _OperationFailure('Write conflict', {'code': 112, 'keyValue': {'token': SECRET}})
+    recorded = "Write conflict, full error: {'code': 112, 'keyValue': 'REDACTED'}"
+    try:
+        raise NetworkTimeout(str(cause)) from cause
+    except NetworkTimeout as error:
+        assert Throwable(error, None).message == recorded
+    assert Throwable(EncryptionError(cause), None).message == recorded
+
+
+def test_pymongo_errors_without_error_document():
+    # e.g. a ServerSelectionTimeoutError with the errors of its servers
+    message = 'No primary available, Topology: <NotPrimaryError("not primary, full error: {\'code\': 10107}")>, ...'
+    error = _OperationFailure(message, None)
+    duplicate_key_error = _OperationFailure('E11000 dup key: { email: "a, full error: %s" }' % SECRET, None)
+
+    assert Throwable(error, None).message == message
+    assert Throwable(duplicate_key_error, None).message == 'E11000 dup key: REDACTED'
+
+
+def test_malformed_stack():
+    throwable = Throwable(Exception('why'), 'a')
+
+    assert throwable.stack is None and throwable.message == 'why'
+
+
+def test_pymongo_error_subclasses():
+    class TenantWriteError(_OperationFailure):
+        __module__ = 'myapp.errors'
+
+    class _Error(Exception):
+        pass
+    _Error.__module__ = None
+
+    error = TenantWriteError('Document failed validation', {'code': 121, 'errInfo': {'consideredValue': SECRET}})
+
+    assert Throwable(error, None).message == (
+        "Document failed validation, full error: {'code': 121, 'errInfo': 'REDACTED'}")
+    assert Throwable(_Error('why'), None).message == 'why'
+
+
+def test_pymongo_errors_with_full_documents(monkeypatch):
+    # e.g. with a pymongo listener that records full documents
+    monkeypatch.setattr(throwable, '_REDACTION_RULES', throwable._REDACTION_RULES)
+    throwable._record_full_pymongo_documents()
+    error = _OperationFailure('Document failed validation', {'code': 121, 'errInfo': {'consideredValue': SECRET}})
+    url_error = Exception('Max retries exceeded with url: /key?token=' + SECRET)
+
+    assert SECRET in Throwable(error, None).message
+    assert Throwable(url_error, None).message == 'Max retries exceeded with url: /key?token=REDACTED'
+
+
+def test_url_query_values():
+    assert Throwable(Exception('Max retries exceeded with url: /key?token=' + SECRET), None).message == (
+        'Max retries exceeded with url: /key?token=REDACTED')
+
+
+def test_details_of_other_errors():
+    class _APIError(Exception):
+        @property
+        def details(self):
+            raise KeyError('details')
+
+    assert Throwable(_APIError('upstream failed with 503, full error: none'), None).message == (
+        'upstream failed with 503, full error: none')
+
+
+def test_redaction_rules(monkeypatch):
+    def redact_token(exception, message):
+        return message.replace(SECRET, 'REDACTED')
+
+    monkeypatch.setattr(throwable, '_REDACTION_RULES', throwable._REDACTION_RULES + (redact_token,))
+
+    assert Throwable(Exception('token ' + SECRET), None).message == 'token REDACTED'
+
+
+def test_message_not_recorded_when_redaction_fails(monkeypatch):
+    def fail(exception, message):
+        raise ValueError()
+
+    monkeypatch.setattr(throwable, '_REDACTION_RULES', (fail,))
+
+    assert not hasattr(Throwable(Exception('token ' + SECRET), None), 'message')