@SuppressWarnings("unchecked")
        @Override
        public void mergeFrom(long epoch, Delta source) {
            HashTier<T> other = (HashTier<T>) source;
            // As an optimization, the deltaTable might not exist for a new key
            // as there is no previous value
            if (other.deltaTable != null) {
                List<T> list = new ArrayList<>();
                Object[] otherElements = other.deltaTable.baseElements();
                for (int slot = 0; slot < otherElements.length; slot++) {
                    BaseHashTable.unpackSlot(list, otherElements, slot);
                    for (T element : list) {
                        // When merging in a later hash tier, we want to keep only the elements
                        // that were present at our epoch.
                        if (element.startEpoch() <= epoch) {
                            if (deltaTable == null) {
                                deltaTable = new BaseHashTable<>(1);
                            }
                            deltaTable.baseAddOrReplace(element);
                        }
                    }
                }
            }
        }
A special epoch value that represents the latest data.
server-common/src/main/java/org/apache/kafka/timeline/SnapshottableHashTable.java