diff --git a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/FunctionCache.java b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/FunctionCache.java
new file mode 100644
index 00000000..881cf5a9
--- /dev/null
+++ b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/FunctionCache.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2023 Amazon.com, Inc. or its affiliates.
+ * 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 software.amazon.kinesis.common;
+
+import java.util.function.Function;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Caches the result from a {@link Function}. Caching is especially useful when
+ * invoking the function is an expensive call that produces a reusable result.
+ * If the input value should be fixed, {@link SupplierCache} may be used.
+ * <br/><br/>
+ * Note that if {@code f(x)=X} is cached, {@code X} will be returned for every
+ * successive query of this cache regardless of the input parameter. This is
+ * by design under the assumption that {@code X} is a viable response for
+ * other invocations.
+ *
+ * @param <IN> input type
+ * @param <OUT> output type
+ */
+@RequiredArgsConstructor
+public class FunctionCache<IN, OUT> extends SynchronizedCache<OUT> {
+
+    private final Function<IN, OUT> function;
+
+    /**
+     * Returns the cached result. If the cache is null, the function will be
+     * invoked to populate the cache.
+     *
+     * @param input input argument to the underlying function
+     * @return cached result which may be null
+     */
+    public OUT get(final IN input) {
+        return get(() -> function.apply(input));
+    }
+
+}
diff --git a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SupplierCache.java b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SupplierCache.java
index 632e4b8f..72d75b92 100644
--- a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SupplierCache.java
+++ b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SupplierCache.java
@@ -9,12 +9,10 @@ import lombok.RequiredArgsConstructor;
  * {@link Supplier#get()} is an expensive call that produces static results.
  */
 @RequiredArgsConstructor
-public class SupplierCache<T> {
+public class SupplierCache<T> extends SynchronizedCache<T> {
 
     private final Supplier<T> supplier;
 
-    private volatile T result;
-
     /**
      * Returns the cached result. If the cache is null, the supplier will be
      * invoked to populate the cache.
@@ -22,15 +20,7 @@ public class SupplierCache<T> {
      * @return cached result which may be null
      */
     public T get() {
-        if (result == null) {
-            synchronized (this) {
-                // double-check lock
-                if (result == null) {
-                    result = supplier.get();
-                }
-            }
-        }
-        return result;
+        return get(supplier);
     }
 
 }
diff --git a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SynchronizedCache.java b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SynchronizedCache.java
new file mode 100644
index 00000000..3df241d3
--- /dev/null
+++ b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/SynchronizedCache.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2023 Amazon.com, Inc. or its affiliates.
+ * 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 software.amazon.kinesis.common;
+
+import java.util.function.Supplier;
+
+/**
+ * A synchronized, "no frills" cache that preserves the first non-null value
+ * returned from a {@link Supplier}.
+ *
+ * @param <R> result type
+ */
+public class SynchronizedCache<R> {
+
+    private volatile R result;
+
+    /**
+     * Returns the cached result. If the cache is null, the supplier will be
+     * invoked to populate the cache.
+     *
+     * @param supplier supplier to invoke if the cache is null
+     * @return cached result which may be null
+     */
+    protected R get(final Supplier<R> supplier) {
+        if (result == null) {
+            synchronized (this) {
+                // double-check lock
+                if (result == null) {
+                    result = supplier.get();
+                }
+            }
+        }
+        return result;
+    }
+
+}
diff --git a/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/FunctionCacheTest.java b/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/FunctionCacheTest.java
new file mode 100644
index 00000000..2f55af4b
--- /dev/null
+++ b/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/FunctionCacheTest.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2023 Amazon.com, Inc. or its affiliates.
+ * 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 software.amazon.kinesis.common;
+
+import java.util.function.Function;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class FunctionCacheTest {
+
+    @Mock
+    private Function<Integer, Object> mockFunction;
+
+    private FunctionCache<Integer, Object> cache;
+
+    @Before
+    public void setUp() {
+        cache = new FunctionCache<>(mockFunction);
+    }
+
+    /**
+     * Test that the cache stops invoking the encapsulated {@link Function}
+     * after it returns a non-null value.
+     */
+    @Test
+    public void testCache() {
+        final int expectedValue = 3;
+        when(mockFunction.apply(expectedValue)).thenReturn(expectedValue);
+
+        assertNull(cache.get(1));
+        assertNull(cache.get(2));
+        assertEquals(expectedValue, cache.get(3));
+        assertEquals(expectedValue, cache.get(4));
+        assertEquals(expectedValue, cache.get(5));
+        verify(mockFunction, times(expectedValue)).apply(anyInt());
+    }
+}
\ No newline at end of file
diff --git a/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/SynchronizedCacheTest.java b/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/SynchronizedCacheTest.java
new file mode 100644
index 00000000..bad3f3cf
--- /dev/null
+++ b/amazon-kinesis-client/src/test/java/software/amazon/kinesis/common/SynchronizedCacheTest.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2023 Amazon.com, Inc. or its affiliates.
+ * 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 software.amazon.kinesis.common;
+
+import java.util.function.Supplier;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SynchronizedCacheTest {
+
+    private static final Object DUMMY_RESULT = SynchronizedCacheTest.class;
+
+    @Mock
+    private Supplier<Object> mockSupplier;
+
+    private final SynchronizedCache<Object> cache = new SynchronizedCache<>();
+
+    @Test
+    public void testCache() {
+        when(mockSupplier.get()).thenReturn(DUMMY_RESULT);
+
+        final Object result1 = cache.get(mockSupplier);
+        final Object result2 = cache.get(mockSupplier);
+
+        assertEquals(DUMMY_RESULT, result1);
+        assertSame(result1, result2);
+        verify(mockSupplier).get();
+    }
+
+    @Test
+    public void testCacheWithNullResult() {
+        when(mockSupplier.get()).thenReturn(null).thenReturn(DUMMY_RESULT);
+
+        assertNull(cache.get(mockSupplier));
+        assertEquals(DUMMY_RESULT, cache.get(mockSupplier));
+        assertEquals(DUMMY_RESULT, cache.get(mockSupplier));
+        verify(mockSupplier, times(2)).get();
+    }
+}
\ No newline at end of file