def METHOD_NAME(self, offsets: MutableMapping[TP, int]) -> None:
    """Call when committing source topic partitions."""
    # flush any pending persisted offsets added by
    # persist_offset_on_commit
    for tp in offsets:
        self.on_commit_tp(tp)