def METHOD_NAME(
    self,
    batch: Iterable[EventT],
    to_key: Callable[[Any], Any],
    to_value: Callable[[Any], Any],
) -> None:
    """Apply batch of changelog events to local state."""
    tp_offsets: Dict[TP, int] = {}
    for event in batch:
        tp, offset = event.message.tp, event.message.offset
        tp_offsets[tp] = (
            offset if tp not in tp_offsets else max(offset, tp_offsets[tp])
        )
        if event.key is None:
            raise RuntimeError("Changelog key cannot be None")
        operation, key = event.key
        key = to_key(key)
        value: Any = to_value(event.value)
        self[key].apply_changelog_event(operation, value)
    for tp, offset in tp_offsets.items():
        self.set_persisted_offset(tp, offset)