diff --git a/bson-record-codec/src/main/org/bson/codecs/record/RecordCodec.java b/bson-record-codec/src/main/org/bson/codecs/record/RecordCodec.java
index d0d2eae5b8f..050ab14ce30 100644
--- a/bson-record-codec/src/main/org/bson/codecs/record/RecordCodec.java
+++ b/bson-record-codec/src/main/org/bson/codecs/record/RecordCodec.java
@@ -89,12 +89,11 @@ Object getValue(final Record record) throws InvocationTargetException, IllegalAc
         @SuppressWarnings("deprecation")
         private static Codec<?> computeCodec(final List<Type> typeParameters, final RecordComponent component,
                 final CodecRegistry codecRegistry) {
-            var codec = codecRegistry.get(toWrapper(resolveComponentType(typeParameters, component)));
-            if (codec instanceof Parameterizable parameterizableCodec
-                    && component.getGenericType() instanceof ParameterizedType parameterizedType) {
-                codec = parameterizableCodec.parameterize(codecRegistry,
-                        resolveActualTypeArguments(typeParameters, component.getDeclaringRecord(), parameterizedType));
-            }
+            var rawType = toWrapper(resolveComponentType(typeParameters, component));
+            var codec = component.getGenericType() instanceof ParameterizedType parameterizedType
+                    ? codecRegistry.get(rawType,
+                    resolveActualTypeArguments(typeParameters, component.getDeclaringRecord(), parameterizedType))
+                    : codecRegistry.get(rawType);
             BsonType bsonRepresentationType = null;
 
             if (component.isAnnotationPresent(BsonRepresentation.class)) {
@@ -271,7 +270,7 @@ private static <T extends Annotation> void validateAnnotationOnlyOnField(final R
     }
 
     RecordCodec(final Class<T> clazz, final CodecRegistry codecRegistry, final List<Type> types) {
-        if (types.size() != clazz.getTypeParameters().length) {
+        if (types.size() != clazz.getTypeParameters().length || types.isEmpty()) {
             throw new CodecConfigurationException("Unexpected number of type parameters for record class " + clazz);
         }
         this.clazz = notNull("class", clazz);
diff --git a/bson-record-codec/src/test/unit/org/bson/codecs/record/RecordCodecTest.java b/bson-record-codec/src/test/unit/org/bson/codecs/record/RecordCodecTest.java
index 9731b96e8a9..a21fbd769ad 100644
--- a/bson-record-codec/src/test/unit/org/bson/codecs/record/RecordCodecTest.java
+++ b/bson-record-codec/src/test/unit/org/bson/codecs/record/RecordCodecTest.java
@@ -50,6 +50,8 @@
 import org.bson.codecs.record.samples.TestRecordWithNestedParameterizedRecord;
 import org.bson.codecs.record.samples.TestRecordWithParameterizedRecord;
 import org.bson.codecs.record.samples.TestRecordWithPojoAnnotations;
+import org.bson.codecs.record.samples.TestSelfReferentialHolderRecord;
+import org.bson.codecs.record.samples.TestSelfReferentialRecord;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
 import org.junit.jupiter.api.Test;
@@ -341,6 +343,36 @@ public void testRecordWithExtraData() {
         assertEquals(testRecord, decoded);
     }
 
+    @Test
+    public void testSelfReferentialRecords() {
+        var registry = fromProviders(new RecordCodecProvider(), Bson.DEFAULT_CODEC_REGISTRY);
+        var codec = registry.get(TestSelfReferentialHolderRecord.class);
+        var testRecord = new TestSelfReferentialHolderRecord("0",
+                new TestSelfReferentialRecord<>("1",
+                new TestSelfReferentialRecord<>("2", null, null),
+                new TestSelfReferentialRecord<>("3", null, null)));
+
+        var document = new BsonDocument();
+
+        // when
+        codec.encode(new BsonDocumentWriter(document), testRecord, EncoderContext.builder().build());
+
+        // then
+        assertEquals(
+                new BsonDocument("_id", new BsonString("0"))
+                        .append("selfReferentialRecord",
+                        new BsonDocument("name", new BsonString("1"))
+                        .append("left", new BsonDocument("name", new BsonString("2")))
+                        .append("right", new BsonDocument("name", new BsonString("3")))),
+                document);
+
+        // when
+        var decoded = codec.decode(new BsonDocumentReader(document), DecoderContext.builder().build());
+
+        // then
+        assertEquals(testRecord, decoded);
+    }
+
     @Test
     public void testExceptionsForAnnotationsNotOnRecordComponent() {
         assertThrows(CodecConfigurationException.class, () ->
diff --git a/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialHolderRecord.java b/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialHolderRecord.java
new file mode 100644
index 00000000000..557243d3f50
--- /dev/null
+++ b/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialHolderRecord.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.bson.codecs.record.samples;
+
+import org.bson.codecs.pojo.annotations.BsonId;
+
+public record TestSelfReferentialHolderRecord(@BsonId String id,
+                                              TestSelfReferentialRecord<String> selfReferentialRecord) {
+}
diff --git a/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialRecord.java b/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialRecord.java
new file mode 100644
index 00000000000..5f097854670
--- /dev/null
+++ b/bson-record-codec/src/test/unit/org/bson/codecs/record/samples/TestSelfReferentialRecord.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.bson.codecs.record.samples;
+
+import com.mongodb.lang.Nullable;
+
+public record TestSelfReferentialRecord<T>(T name,
+                                        @Nullable TestSelfReferentialRecord<T> left,
+                                        @Nullable TestSelfReferentialRecord<T> right) {
+}
diff --git a/bson/src/main/org/bson/assertions/Assertions.java b/bson/src/main/org/bson/assertions/Assertions.java
index e3de9f1eaf1..31b81a45f50 100644
--- a/bson/src/main/org/bson/assertions/Assertions.java
+++ b/bson/src/main/org/bson/assertions/Assertions.java
@@ -17,6 +17,8 @@
 
 package org.bson.assertions;
 
+import javax.annotation.Nullable;
+
 /**
  * <p>Design by contract assertions.</p> <p>This class is not part of the public API and may be removed or changed at any time.</p>
  */
@@ -82,6 +84,38 @@ public static <T> T isTrueArgument(final String name, final T value, final boole
         return value;
     }
 
+    /**
+     * @return Never completes normally. The return type is {@link AssertionError} to allow writing {@code throw fail()}.
+     * This may be helpful in non-{@code void} methods.
+     * @throws AssertionError Always
+     */
+    public static AssertionError fail() throws AssertionError {
+        throw new AssertionError();
+    }
+
+    /**
+     * @param msg The failure message.
+     * @return Never completes normally. The return type is {@link AssertionError} to allow writing {@code throw fail("failure message")}.
+     * This may be helpful in non-{@code void} methods.
+     * @throws AssertionError Always
+     */
+    public static AssertionError fail(final String msg) throws AssertionError {
+        throw new AssertionError(assertNotNull(msg));
+    }
+
+    /**
+     * @param value A value to check.
+     * @param <T>   The type of {@code value}.
+     * @return {@code value}
+     * @throws AssertionError If {@code value} is {@code null}.
+     */
+    public static <T> T assertNotNull(@Nullable final T value) throws AssertionError {
+        if (value == null) {
+            throw new AssertionError();
+        }
+        return value;
+    }
+
     /**
      * Cast an object to the given class and return it, or throw IllegalArgumentException if it's not assignable to that class.
      *
diff --git a/bson/src/main/org/bson/codecs/ContainerCodecHelper.java b/bson/src/main/org/bson/codecs/ContainerCodecHelper.java
index 6f9573ffe02..5969763546b 100644
--- a/bson/src/main/org/bson/codecs/ContainerCodecHelper.java
+++ b/bson/src/main/org/bson/codecs/ContainerCodecHelper.java
@@ -71,12 +71,7 @@ static Codec<?> getCodec(final CodecRegistry codecRegistry, final Type type) {
             return codecRegistry.get((Class<?>) type);
         } else if (type instanceof ParameterizedType) {
             ParameterizedType parameterizedType = (ParameterizedType) type;
-            Codec<?> rawCodec = codecRegistry.get((Class<?>) parameterizedType.getRawType());
-            if (rawCodec instanceof Parameterizable) {
-                return ((Parameterizable) rawCodec).parameterize(codecRegistry, Arrays.asList(parameterizedType.getActualTypeArguments()));
-            } else {
-                return rawCodec;
-            }
+            return codecRegistry.get((Class<?>) parameterizedType.getRawType(), Arrays.asList(parameterizedType.getActualTypeArguments()));
         } else {
             throw new CodecConfigurationException("Unsupported generic type of container: " + type);
         }
diff --git a/bson/src/main/org/bson/codecs/Parameterizable.java b/bson/src/main/org/bson/codecs/Parameterizable.java
index 0b6f219d9da..917801b891c 100644
--- a/bson/src/main/org/bson/codecs/Parameterizable.java
+++ b/bson/src/main/org/bson/codecs/Parameterizable.java
@@ -19,6 +19,7 @@
 import org.bson.codecs.configuration.CodecRegistry;
 
 import java.lang.reflect.Type;
+import java.util.Collection;
 import java.util.List;
 
 /**
@@ -31,7 +32,10 @@ public interface Parameterizable {
      * Recursively parameterize the codec with the given registry and generic type arguments.
      *
      * @param codecRegistry the code registry to use to resolve codecs for the generic type arguments
-     * @param types the types that are parameterizing the containing type.
+     * @param types         the types that are parameterizing the containing type.  The size of the list should be equal to the number of type
+     *                      parameters of the class whose codec is being parameterized, e.g. for a {@link Collection} the size of the list
+     *                      would be one since {@code Collection} has a single type parameter. Additionally, the size will never be 0
+     *                      since there is no purpose in parameterizing a codec for a type that has no type parameters.
      * @return the Codec parameterized with the given types
      */
     Codec<?> parameterize(CodecRegistry codecRegistry, List<Type> types);
diff --git a/bson/src/main/org/bson/codecs/configuration/CodecRegistry.java b/bson/src/main/org/bson/codecs/configuration/CodecRegistry.java
index c7edf9aade4..31479d31820 100644
--- a/bson/src/main/org/bson/codecs/configuration/CodecRegistry.java
+++ b/bson/src/main/org/bson/codecs/configuration/CodecRegistry.java
@@ -16,8 +16,12 @@
 
 package org.bson.codecs.configuration;
 
+import org.bson.assertions.Assertions;
 import org.bson.codecs.Codec;
 
+import java.lang.reflect.Type;
+import java.util.List;
+
 /**
  * A registry of Codec instances searchable by the class that the Codec can encode and decode.
  *
@@ -28,7 +32,11 @@
  *
  * <p>As of the 4.0 release, this class extends the {@code CodecProvider} interface.  This capability was introduced to enable nesting
  * registries inside another registry.</p>
+ *
+ * <p>Applications are encouraged to NOT implement this interface, but rather use the factory methods in {@link CodecRegistries}.</p>
+ *
  * @since 3.0
+ * @see CodecRegistries
  */
 public interface CodecRegistry extends CodecProvider {
     /**
@@ -40,4 +48,26 @@ public interface CodecRegistry extends CodecProvider {
      * @throws CodecConfigurationException if the registry does not contain a codec for the given class.
      */
     <T> Codec<T> get(Class<T> clazz);
+
+    /**
+     * Gets a Codec for the given parameterized class, after resolving any type variables with the given type arguments.
+     *
+     * <p>
+     * The default behavior is to throw a {@link AssertionError}, as it's expected that {@code CodecRegistry} implementations are always
+     * provided by this library and will override the method appropriately.
+     * </p>
+     *
+     * @param clazz         the parameterized class
+     * @param typeArguments the type arguments to apply to the parameterized class.  This list may be empty but not null.
+     * @param <T>           the class type
+     * @return a codec for the given class, with the given type parameters resolved
+     * @throws CodecConfigurationException if no codec can be found for the given class and type arguments.
+     * @throws AssertionError              by default, if the implementation does not override this method, or if no codec can be found
+     *                                     for the given class and type arguments.
+     * @see org.bson.codecs.Parameterizable
+     * @since 4.8
+     */
+    default <T> Codec<T> get(Class<T> clazz, List<Type> typeArguments) {
+        throw Assertions.fail("This method should have been overridden but was not.");
+    }
 }
diff --git a/bson/src/main/org/bson/internal/ChildCodecRegistry.java b/bson/src/main/org/bson/internal/ChildCodecRegistry.java
index 3e78967accf..8b911f541ee 100644
--- a/bson/src/main/org/bson/internal/ChildCodecRegistry.java
+++ b/bson/src/main/org/bson/internal/ChildCodecRegistry.java
@@ -20,36 +20,64 @@
 import org.bson.codecs.Codec;
 import org.bson.codecs.configuration.CodecRegistry;
 
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.Optional;
+
+import static java.lang.String.format;
+import static org.bson.assertions.Assertions.isTrueArgument;
+import static org.bson.assertions.Assertions.notNull;
+
 // An implementation of CodecRegistry that is used to detect cyclic dependencies between Codecs
 class ChildCodecRegistry<T> implements CodecRegistry {
 
     private final ChildCodecRegistry<?> parent;
     private final CycleDetectingCodecRegistry registry;
     private final Class<T> codecClass;
+    private final List<Type> types;
 
-    ChildCodecRegistry(final CycleDetectingCodecRegistry registry, final Class<T> codecClass) {
+    ChildCodecRegistry(final CycleDetectingCodecRegistry registry, final Class<T> codecClass, final List<Type> types) {
         this.codecClass = codecClass;
         this.parent = null;
         this.registry = registry;
+        this.types = types;
     }
 
-
-    private ChildCodecRegistry(final ChildCodecRegistry<?> parent, final Class<T> codecClass) {
+    private ChildCodecRegistry(final ChildCodecRegistry<?> parent, final Class<T> codecClass, final List<Type> types) {
         this.parent = parent;
         this.codecClass = codecClass;
         this.registry = parent.registry;
+        this.types = types;
     }
 
     public Class<T> getCodecClass() {
         return codecClass;
     }
 
+    public Optional<List<Type>> getTypes() {
+        return Optional.ofNullable(types);
+    }
+
     // Gets a Codec, but if it detects a cyclic dependency, return a LazyCodec which breaks the chain.
     public <U> Codec<U> get(final Class<U> clazz) {
         if (hasCycles(clazz)) {
-            return new LazyCodec<U>(registry, clazz);
+            return new LazyCodec<>(registry, clazz, null);
+        } else {
+            return registry.get(new ChildCodecRegistry<>(this, clazz, null));
+        }
+    }
+
+    @Override
+    public <U> Codec<U> get(final Class<U> clazz, final List<Type> typeArguments) {
+        notNull("typeArguments", typeArguments);
+        isTrueArgument("typeArguments is not empty", !typeArguments.isEmpty());
+        isTrueArgument(format("typeArguments size should equal the number of type parameters in class %s, but is %d",
+                        clazz, typeArguments.size()),
+                clazz.getTypeParameters().length == typeArguments.size());
+        if (hasCycles(clazz)) {
+            return new LazyCodec<U>(registry, clazz, typeArguments);
         } else {
-            return registry.get(new ChildCodecRegistry<U>(this, clazz));
+            return registry.get(new ChildCodecRegistry<>(this, clazz, typeArguments));
         }
     }
 
diff --git a/bson/src/main/org/bson/internal/CodecCache.java b/bson/src/main/org/bson/internal/CodecCache.java
index 12ea83ffcea..c4b59d713b4 100644
--- a/bson/src/main/org/bson/internal/CodecCache.java
+++ b/bson/src/main/org/bson/internal/CodecCache.java
@@ -19,6 +19,9 @@
 import org.bson.codecs.Codec;
 import org.bson.codecs.configuration.CodecConfigurationException;
 
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -26,29 +29,65 @@
 import static java.lang.String.format;
 
 final class CodecCache {
-    private final ConcurrentMap<Class<?>, Optional<Codec<?>>> codecCache = new ConcurrentHashMap<>();
 
-    public boolean containsKey(final Class<?> clazz) {
-        return codecCache.containsKey(clazz);
+    static final class CodecCacheKey {
+        private final Class<?> clazz;
+        private final List<Type> types;
+
+        CodecCacheKey(final Class<?> clazz, final List<Type> types) {
+            this.clazz = clazz;
+            this.types = types;
+        }
+
+        @Override
+        public boolean equals(final Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            CodecCacheKey that = (CodecCacheKey) o;
+            return clazz.equals(that.clazz) && Objects.equals(types, that.types);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(clazz, types);
+        }
+
+        @Override
+        public String toString() {
+            return "CodecCacheKey{"
+                    + "clazz=" + clazz
+                    + ", types=" + types
+                    + '}';
+        }
+    }
+
+    private final ConcurrentMap<CodecCacheKey, Optional<Codec<?>>> codecCache = new ConcurrentHashMap<>();
+
+    public boolean containsKey(final CodecCacheKey codecCacheKey) {
+        return codecCache.containsKey(codecCacheKey);
     }
 
-    public void put(final Class<?> clazz, final Codec<?> codec){
-        codecCache.put(clazz, Optional.ofNullable(codec));
+    public void put(final CodecCacheKey codecCacheKey, final Codec<?> codec){
+        codecCache.put(codecCacheKey, Optional.ofNullable(codec));
     }
 
     @SuppressWarnings("unchecked")
-    public synchronized <T> Codec<T> putIfMissing(final Class<T> clazz, final Codec<T> codec) {
-        Optional<Codec<?>> cachedCodec = codecCache.computeIfAbsent(clazz, clz -> Optional.of(codec));
+    public synchronized <T> Codec<T> putIfMissing(final CodecCacheKey codecCacheKey, final Codec<T> codec) {
+        Optional<Codec<?>> cachedCodec = codecCache.computeIfAbsent(codecCacheKey, clz -> Optional.of(codec));
         if (cachedCodec.isPresent()) {
             return (Codec<T>) cachedCodec.get();
         }
-        codecCache.put(clazz, Optional.of(codec));
+        codecCache.put(codecCacheKey, Optional.of(codec));
         return codec;
     }
 
     @SuppressWarnings("unchecked")
-    public <T> Codec<T> getOrThrow(final Class<T> clazz) {
-        return (Codec<T>) codecCache.getOrDefault(clazz, Optional.empty()).orElseThrow(
-                () -> new CodecConfigurationException(format("Can't find a codec for %s.", clazz)));
+    public <T> Codec<T> getOrThrow(final CodecCacheKey codecCacheKey) {
+        return (Codec<T>) codecCache.getOrDefault(codecCacheKey, Optional.empty()).orElseThrow(
+                () -> new CodecConfigurationException(format("Can't find a codec for %s.", codecCacheKey)));
     }
 }
diff --git a/bson/src/main/org/bson/internal/LazyCodec.java b/bson/src/main/org/bson/internal/LazyCodec.java
index e8c7b67adf9..0e7f94e9441 100644
--- a/bson/src/main/org/bson/internal/LazyCodec.java
+++ b/bson/src/main/org/bson/internal/LazyCodec.java
@@ -23,14 +23,19 @@
 import org.bson.codecs.EncoderContext;
 import org.bson.codecs.configuration.CodecRegistry;
 
+import java.lang.reflect.Type;
+import java.util.List;
+
 class LazyCodec<T> implements Codec<T> {
     private final CodecRegistry registry;
     private final Class<T> clazz;
+    private final List<Type> types;
     private volatile Codec<T> wrapped;
 
-    LazyCodec(final CodecRegistry registry, final Class<T> clazz) {
+    LazyCodec(final CodecRegistry registry, final Class<T> clazz, final List<Type> types) {
         this.registry = registry;
         this.clazz = clazz;
+        this.types = types;
     }
 
     @Override
@@ -50,7 +55,11 @@ public T decode(final BsonReader reader, final DecoderContext decoderContext) {
 
     private Codec<T> getWrapped() {
         if (wrapped == null) {
-            wrapped = registry.get(clazz);
+            if (types == null) {
+                wrapped = registry.get(clazz);
+            } else {
+                wrapped = registry.get(clazz, types);
+            }
         }
 
         return wrapped;
diff --git a/bson/src/main/org/bson/internal/ProvidersCodecRegistry.java b/bson/src/main/org/bson/internal/ProvidersCodecRegistry.java
index 8800131a282..10bcb6d2c90 100644
--- a/bson/src/main/org/bson/internal/ProvidersCodecRegistry.java
+++ b/bson/src/main/org/bson/internal/ProvidersCodecRegistry.java
@@ -17,13 +17,18 @@
 package org.bson.internal;
 
 import org.bson.codecs.Codec;
+import org.bson.codecs.Parameterizable;
 import org.bson.codecs.configuration.CodecProvider;
 import org.bson.codecs.configuration.CodecRegistry;
+import org.bson.internal.CodecCache.CodecCacheKey;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
 
+import static java.lang.String.format;
 import static org.bson.assertions.Assertions.isTrueArgument;
+import static org.bson.assertions.Assertions.notNull;
 
 public final class ProvidersCodecRegistry implements CodecRegistry, CycleDetectingCodecRegistry {
     private final List<CodecProvider> codecProviders;
@@ -36,7 +41,17 @@ public ProvidersCodecRegistry(final List<? extends CodecProvider> codecProviders
 
     @Override
     public <T> Codec<T> get(final Class<T> clazz) {
-        return get(new ChildCodecRegistry<T>(this, clazz));
+        return get(new ChildCodecRegistry<T>(this, clazz, null));
+    }
+
+    @Override
+    public <T> Codec<T> get(final Class<T> clazz, final List<Type> typeArguments) {
+        notNull("typeArguments", typeArguments);
+        isTrueArgument("typeArguments is not empty", !typeArguments.isEmpty());
+        isTrueArgument(format("typeArguments size should equal the number of type parameters in class %s, but is %d",
+                        clazz, typeArguments.size()),
+                clazz.getTypeParameters().length == typeArguments.size());
+        return get(new ChildCodecRegistry<T>(this, clazz, typeArguments));
     }
 
     @SuppressWarnings("rawtypes")
@@ -50,18 +65,22 @@ public <T> Codec<T> get(final Class<T> clazz, final CodecRegistry registry) {
         return null;
     }
 
-    @SuppressWarnings({"rawtypes" })
+    @SuppressWarnings({"unchecked"})
     public <T> Codec<T> get(final ChildCodecRegistry<T> context) {
-        if (!codecCache.containsKey(context.getCodecClass())) {
+        CodecCacheKey codecCacheKey = new CodecCacheKey(context.getCodecClass(), context.getTypes().orElse(null));
+        if (!codecCache.containsKey(codecCacheKey)) {
             for (CodecProvider provider : codecProviders) {
                 Codec<T> codec = provider.get(context.getCodecClass(), context);
                 if (codec != null) {
-                    return codecCache.putIfMissing(context.getCodecClass(), codec);
+                    if (codec instanceof Parameterizable && context.getTypes().isPresent()) {
+                        codec = (Codec<T>) ((Parameterizable) codec).parameterize(context, context.getTypes().get());
+                    }
+                    return codecCache.putIfMissing(codecCacheKey, codec);
                 }
             }
-            codecCache.put(context.getCodecClass(), null);
+            codecCache.put(codecCacheKey, null);
         }
-        return codecCache.getOrThrow(context.getCodecClass());
+        return codecCache.getOrThrow(codecCacheKey);
     }
 
     @Override
diff --git a/bson/src/test/unit/org/bson/internal/CodecCacheSpecification.groovy b/bson/src/test/unit/org/bson/internal/CodecCacheSpecification.groovy
index e4eab0b34a2..5d34d29ff2e 100644
--- a/bson/src/test/unit/org/bson/internal/CodecCacheSpecification.groovy
+++ b/bson/src/test/unit/org/bson/internal/CodecCacheSpecification.groovy
@@ -27,16 +27,39 @@ class CodecCacheSpecification extends Specification {
         when:
         def codec = new MinKeyCodec()
         def cache = new CodecCache()
-        cache.put(MinKey, codec)
+        def cacheKey = new CodecCache.CodecCacheKey(MinKey, null)
+        cache.put(cacheKey, codec)
 
         then:
-        cache.getOrThrow(MinKey).is(codec)
+        cache.getOrThrow(cacheKey).is(codec)
     }
 
     def 'should throw if codec for class does not exist'() {
         when:
         def cache = new CodecCache()
-        cache.getOrThrow(MinKey)
+        def cacheKey = new CodecCache.CodecCacheKey(MinKey, null)
+        cache.getOrThrow(cacheKey)
+
+        then:
+        thrown(CodecConfigurationException)
+    }
+
+    def 'should return the cached codec if a codec for the parameterized class exists'() {
+        when:
+        def codec = new MinKeyCodec()
+        def cache = new CodecCache()
+        def cacheKey = new CodecCache.CodecCacheKey(List, [Integer])
+        cache.put(cacheKey, codec)
+
+        then:
+        cache.getOrThrow(cacheKey).is(codec)
+    }
+
+    def 'should throw if codec for the parameterized class does not exist'() {
+        when:
+        def cache = new CodecCache()
+        def cacheKey = new CodecCache.CodecCacheKey(List, [Integer])
+        cache.getOrThrow(cacheKey)
 
         then:
         thrown(CodecConfigurationException)
diff --git a/bson/src/test/unit/org/bson/internal/Holder.java b/bson/src/test/unit/org/bson/internal/Holder.java
new file mode 100644
index 00000000000..afbf2f3f78c
--- /dev/null
+++ b/bson/src/test/unit/org/bson/internal/Holder.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.bson.internal;
+
+import java.util.Collection;
+
+public class Holder {
+    @SuppressWarnings("VisibilityModifier")
+    public Collection<Collection<Integer>> c;
+}
diff --git a/bson/src/test/unit/org/bson/internal/ProvidersCodecRegistrySpecification.groovy b/bson/src/test/unit/org/bson/internal/ProvidersCodecRegistrySpecification.groovy
index 7aad6cb9c15..efc64f80a67 100644
--- a/bson/src/test/unit/org/bson/internal/ProvidersCodecRegistrySpecification.groovy
+++ b/bson/src/test/unit/org/bson/internal/ProvidersCodecRegistrySpecification.groovy
@@ -25,7 +25,10 @@ import org.bson.ByteBufNIO
 import org.bson.codecs.Codec
 import org.bson.codecs.DecoderContext
 import org.bson.codecs.EncoderContext
+import org.bson.codecs.IntegerCodec
 import org.bson.codecs.MinKeyCodec
+import org.bson.codecs.Parameterizable
+import org.bson.codecs.ValueCodecProvider
 import org.bson.codecs.configuration.CodecConfigurationException
 import org.bson.codecs.configuration.CodecProvider
 import org.bson.codecs.configuration.CodecRegistry
@@ -35,9 +38,12 @@ import org.bson.types.MaxKey
 import org.bson.types.MinKey
 import spock.lang.Specification
 
+import java.lang.reflect.Type
 import java.nio.ByteBuffer
 
 import static java.util.Arrays.asList
+import static org.bson.codecs.ContainerCodecHelper.getCodec
+import static org.bson.codecs.configuration.CodecRegistries.fromCodecs
 
 class ProvidersCodecRegistrySpecification extends Specification {
 
@@ -153,6 +159,93 @@ class ProvidersCodecRegistrySpecification extends Specification {
         expect:
         ((SimpleCodec) provider.get(Simple, registry)).registry.is(registry)
     }
+
+    def 'should parameterize codec'() {
+        given:
+        def registry = new ProvidersCodecRegistry([fromCodecs(new CollectionCodec()), new ValueCodecProvider()]);
+
+        when:
+        def codec = registry.get(Collection, [Integer])
+
+        then:
+        codec instanceof ParameterizedCollectionCodec
+        (codec as ParameterizedCollectionCodec).getCodec() instanceof IntegerCodec
+
+        when:
+        def secondCodec = registry.get(Collection, [Integer])
+
+        then:
+        codec == secondCodec
+    }
+
+    def 'should parameterize codec with cycles'() {
+        given:
+        def registry = new ProvidersCodecRegistry([fromCodecs(new CollectionCodec()), new ValueCodecProvider()]);
+
+        when:
+        def codec = registry.get(Collection, [Holder.getField('c').getGenericType()])
+
+        then:
+        codec instanceof ParameterizedCollectionCodec
+        (codec as ParameterizedCollectionCodec).getCodec() instanceof LazyCodec
+
+        when:
+        def secondCodec = registry.get(Collection, [Holder.getField('c').getGenericType()])
+
+        then:
+        codec == secondCodec
+    }
+}
+
+class CollectionCodec implements Codec<Collection<?>>, Parameterizable {
+
+    @Override
+    Collection<?> decode(BsonReader reader, DecoderContext decoderContext) {
+        throw new UnsupportedOperationException()
+    }
+
+    @Override
+    void encode(BsonWriter writer, Collection<?> value, EncoderContext encoderContext) {
+        throw new UnsupportedOperationException()
+    }
+
+    @Override
+    Class<Collection<?>> getEncoderClass() {
+        Collection
+    }
+
+    @Override
+    Codec<?> parameterize(CodecRegistry codecRegistry, List<Type> types) {
+        new ParameterizedCollectionCodec(getCodec(codecRegistry, types.get(0)))
+    }
+}
+
+class ParameterizedCollectionCodec<T> implements Codec<Collection<T>> {
+
+    private final Codec<T> codec
+
+    ParameterizedCollectionCodec(Codec<T> codec) {
+        this.codec = codec
+    }
+
+    Codec<T> getCodec() {
+        codec
+    }
+
+    @Override
+    Collection<T> decode(BsonReader reader, DecoderContext decoderContext) {
+        throw new UnsupportedOperationException()
+    }
+
+    @Override
+    void encode(BsonWriter writer, Collection<T> value, EncoderContext encoderContext) {
+        throw new UnsupportedOperationException()
+    }
+
+    @Override
+    Class<Collection<T>> getEncoderClass() {
+        Collection
+    }
 }
 
 class SingleCodecProvider implements CodecProvider {