diff --git a/server-common/src/main/java/org/apache/kafka/timeline/SnapshottableHashTable.java b/server-common/src/main/java/org/apache/kafka/timeline/SnapshottableHashTable.java
index 9284d5964c51..2977e0616439 100644
--- a/server-common/src/main/java/org/apache/kafka/timeline/SnapshottableHashTable.java
+++ b/server-common/src/main/java/org/apache/kafka/timeline/SnapshottableHashTable.java
@@ -261,8 +261,10 @@ public boolean hasNext() {
                         int tierSlot = slot >>> shift;
                         BaseHashTable.unpackSlot(temp, deltaTable.baseElements(), tierSlot);
                         for (T object : temp) {
-                            if (BaseHashTable.findSlot(object, topTier.length) == slot) {
-                                ready.add(object);
+                            if (object.startEpoch() <= snapshot.epoch()) {
+                                if (BaseHashTable.findSlot(object, topTier.length) == slot) {
+                                    ready.add(object);
+                                }
                             }
                         }
                         temp.clear();
diff --git a/server-common/src/test/java/org/apache/kafka/timeline/SnapshottableHashTableTest.java b/server-common/src/test/java/org/apache/kafka/timeline/SnapshottableHashTableTest.java
index 01ea35935cdd..8846262c2c5f 100644
--- a/server-common/src/test/java/org/apache/kafka/timeline/SnapshottableHashTableTest.java
+++ b/server-common/src/test/java/org/apache/kafka/timeline/SnapshottableHashTableTest.java
@@ -273,6 +273,21 @@ public void testReset() {
         assertIteratorYields(table.snapshottableIterator(Long.MAX_VALUE));
     }
 
+    @Test
+    public void testIteratorAtOlderEpoch() {
+        SnapshotRegistry registry = new SnapshotRegistry(new LogContext());
+        SnapshottableHashTable<TestElement> table =
+                new SnapshottableHashTable<>(registry, 4);
+        assertNull(table.snapshottableAddOrReplace(E_3B));
+        registry.getOrCreateSnapshot(0);
+        assertNull(table.snapshottableAddOrReplace(E_1A));
+        registry.getOrCreateSnapshot(1);
+        assertEquals(E_1A, table.snapshottableAddOrReplace(E_1B));
+        registry.getOrCreateSnapshot(2);
+        assertEquals(E_1B, table.snapshottableRemove(E_1B));
+        assertIteratorYields(table.snapshottableIterator(1), E_3B, E_1A);
+    }
+
     /**
      * Assert that the given iterator contains the given elements, in any order.
      * We compare using reference equality here, rather than object equality.