def METHOD_NAME(ceph_cluster, **kw):
    """
    CEPH-83571452 RADOS:
    Corrupt snap info of an object and run
    list-inconsistent-snapset
    Steps:
        1. create a replica 3 pool
        2. take few pool snaps with writes on objects b/w every snap
        3. chose NON-PRIMARY osd and bring it down
        4. go to backend and using ceph-object-store tool corrupt the snapset of the object
        5. run deep-scrub on the pg
        6. check rados list-inconsistent-pg <pool>
        7. rados list-inconsistent-snapset <pg>
    Args:
        ceph_cluster (ceph.ceph.Ceph): ceph cluster
    """
    log.info("Running CEPH-83571452")
    log.info(METHOD_NAME.__doc__)
    ceph_nodes = kw.get("ceph_nodes")
    config = kw.get("config")
    mons = []
    role = "client"
    for mnode in ceph_nodes:
        if mnode.role == role:
            mons.append(mnode)
    ctrlr = mons[0]
    log.info("chosing mon {cmon} as ctrlrmon".format(cmon=ctrlr.hostname))
    helper = RadosHelper(ctrlr, config, log)
    """create a replica pool"""
    pname = "snapcorrupt_{rand}".format(rand=random.randint(0, 10000))
    try:
        helper.create_pool(pname, 1)
        log.info("Pool {pname} created".format(pname=pname))
    except Exception:
        log.error("Failed to create pool")
        log.error(traceback.format_exc())
        return 1
    time.sleep(5)
    """Get the target PG,osd for corruption operation"""
    oname = "UNIQUEOBJECT{i}".format(i=random.randint(0, 10000))
    cmd = "osd map {pname} {obj} --format json".format(pname=pname, obj=oname)
    (outbuf, err) = helper.raw_cluster_cmd(cmd)
    log.info(outbuf)
    cmdout = json.loads(outbuf)
    targt_pg = cmdout["pgid"]
    targt_osd_id = cmdout["up"][random.randint(1, 2)]
    """write data and take snaps"""
    putobj = "sudo rados -p {pool} put {obj} {path}".format(
        pool=pname, obj=oname, path="/etc/hosts"
    )
    for i in range(10):
        (out, err) = ctrlr.exec_command(cmd=putobj)
        snapcmd = "sudo rados mksnap -p {pool} {sname}".format(
            pool=pname, sname="snap" + str(i)
        )
        (out, err) = ctrlr.exec_command(cmd=snapcmd)
        log.info("put {obj}, snap {snap}".format(obj=oname, snap="snap" + str(i)))
    """
    Goto destination osd, stop the osd
    use ceph-objectstore-tool to corrupt
    snap info
    """
    #    target_osd = ceph_cluster.get_osd_by_id(targt_osd_id)
    #    target_osd_node = target_osd.node
    target_osd_hostname = ceph_cluster.get_osd_metadata(targt_osd_id).get("hostname")
    log.info(target_osd_hostname)
    target_osd_node = ceph_cluster.get_node_by_hostname(target_osd_hostname)
    cot_environment = target_osd_node
    osd_service = ceph_cluster.get_osd_service_name(targt_osd_id)
    partition_path = ceph_cluster.get_osd_metadata(targt_osd_id).get("osd_data")
    helper.kill_osd(target_osd_node, osd_service)
    time.sleep(10)
    osd_metadata = ceph_cluster.get_osd_metadata(targt_osd_id)
    osd_data = osd_metadata.get("osd_data")
    osd_journal = osd_metadata.get("osd_journal")
    if ceph_cluster.containerized:
        docker_image_string = "{docker_registry}/{docker_image}:{docker_tag}".format(
            docker_registry=ceph_cluster.ansible_config.get("ceph_docker_registry"),
            docker_image=ceph_cluster.ansible_config.get("ceph_docker_image"),
            docker_tag=ceph_cluster.ansible_config.get("ceph_docker_image_tag"),
        )
        cot_environment = helper.get_mgr_proxy_container(
            target_osd_node, docker_image_string
        )
        device_mount_data, err = cot_environment.exec_command(
            cmd='mount | grep "{partition_path} "'.format(
                partition_path=partition_path
            ),
            check_ec=False,
        )
        if not device_mount_data:
            cot_environment.exec_command(
                cmd="sudo mount {partition_path} {directory}".format(
                    partition_path=partition_path, directory=osd_data
                )
            )
    slist_cmd = "sudo ceph-objectstore-tool --data-path \
            {osd_data} --journal-path \
            {osd_journal} \
            --head --op list {obj}".format(
        osd_data=osd_data, osd_journal=osd_journal, obj=oname
    )
    (outbuf, err) = cot_environment.exec_command(cmd=slist_cmd)
    log.info(outbuf)
    corrupt_cmd = "sudo ceph-objectstore-tool --data-path \
            {osd_data} --journal-path \
            {osd_journal} \
            {outbuf} clear-snapset corrupt".format(
        osd_data=osd_data, osd_journal=osd_journal, outbuf="'" + (outbuf) + "'"
    )
    (outbuf, err) = cot_environment.exec_command(cmd=corrupt_cmd)
    log.info(outbuf)
    helper.revive_osd(target_osd_node, osd_service)
    time.sleep(10)
    run_scrub = "pg deep-scrub {pgid}".format(pgid=targt_pg)
    (outbuf, err) = helper.raw_cluster_cmd(run_scrub)
    log.info(outbuf)
    while "HEALTH_ERR" and "active+clean+inconsistent" not in outbuf:
        status = "-s --format json"
        (outbuf, err) = helper.raw_cluster_cmd(status)
    log.info("HEALTH_ERR found as expected")
    log.info("inconsistent foud as expected")
    timeout = 300
    found = 0
    while timeout:
        incon_pg = "sudo rados list-inconsistent-pg {pname}".format(pname=pname)
        (outbuf, err) = ctrlr.exec_command(cmd=incon_pg)
        log.info(outbuf)
        if targt_pg not in outbuf:
            time.sleep(1)
            timeout = timeout - 1
        else:
            found = 1
            break
    if timeout == 0 and found == 0:
        log.error("pg not listed as inconsistent")
        return 1
    timeout = 300
    found = 0
    # need to check for inconsistent obj before checking for -
    # - inconsistent snapset in case of snapset corruption on primary osds
    while timeout:
        incon_snap = "sudo rados list-inconsistent-obj {pg}".format(pg=targt_pg)
        (outbuf, err) = ctrlr.exec_command(cmd=incon_snap)
        log.info(outbuf)
        if oname not in outbuf:
            time.sleep(1)
            timeout = timeout - 1
        else:
            found = 1
            break
    if timeout == 0 and found == 0:
        log.error("object is not listed in inconsistent obj/snap")
        return 1
    return 0