text: import logging
  import time
  
  from ceph.parallel import parallel
  
  log = logging.getLogger(__name__)
  
  
  def run(**kw):
      try:
          log.info("Starting CEPH-9471")
          mirror1 = kw.get('test_data')['mirror1']
          mirror2 = kw.get('test_data')['mirror2']
          config = kw.get('config')
          poolname = mirror1.random_string() + '9471pool'
          imagename = mirror1.random_string() + '9471image'
          imagespec = poolname + '/' + imagename
          state_after_demote = 'up+stopped' if mirror1.ceph_version < 3 else 'up+unknown'
  
          mirror1.create_pool(poolname=poolname)
          mirror2.create_pool(poolname=poolname)
          mirror1.create_image(imagespec=imagespec, size=config.get('imagesize'))
          mirror1.config_mirror(mirror2, poolname=poolname, mode='pool')
          mirror2.wait_for_status(poolname=poolname, images_pattern=1)
          mirror1.benchwrite(imagespec=imagespec, io=config.get('io-total'))
          mirror2.wait_for_replay_complete(imagespec=imagespec)
          mirror1.demote(imagespec=imagespec)
          mirror1.wait_for_status(imagespec=imagespec, state_pattern=state_after_demote)
          mirror2.wait_for_status(imagespec=imagespec, state_pattern=state_after_demote)
          with parallel() as p:
              for node in mirror1.ceph_nodes:
                  p.spawn(mirror1.exec_cmd, ceph_args=False, cmd='reboot',
                          node=node, check_ec=False)
          mirror2.promote(imagespec=imagespec)
          mirror2.benchwrite(imagespec=imagespec, io=config.get('io-total'))
          time.sleep(30)
          mirror2.check_data(peercluster=mirror1, imagespec=imagespec)
          mirror2.demote(imagespec=imagespec)
          mirror2.wait_for_status(imagespec=imagespec, state_pattern=state_after_demote)
          mirror1.wait_for_status(imagespec=imagespec, state_pattern=state_after_demote)
          mirror1.promote(imagespec=imagespec)
          mirror1.benchwrite(imagespec=imagespec, io=config.get('io-total'))
          mirror1.check_data(peercluster=mirror2, imagespec=imagespec)
          mirror1.clean_up(peercluster=mirror2, pools=[poolname])
          return 0
  
      except Exception as e:
          log.exception(e)
          return 1
  