def execute_plan(self, plan, allow_rf_change=False):
        """Submit reassignment plan for execution."""
        reassignment_path = '{admin}/{reassignment_node}'\
            .format(admin=ADMIN_PATH, reassignment_node=REASSIGNMENT_NODE)
        plan_json = dump_json(plan)
        base_plan = self.get_cluster_plan()
        if not validate_plan(plan, base_plan, allow_rf_change=allow_rf_change):
            _log.error('Given plan is invalid. Aborting new reassignment plan ... {plan}'.format(plan=plan))
            return False
        # Send proposed-plan to zookeeper
        try:
            _log.info('Sending plan to Zookeeper...')
            self.create(reassignment_path, plan_json, makepath=True)
            _log.info(
                'Re-assign partitions node in Zookeeper updated successfully '
                'with {plan}'.format(plan=plan),
            )
            return True
        except NodeExistsError:
            _log.warning('Previous plan in progress. Exiting..')
            _log.warning('Aborting new reassignment plan... {plan}'.format(plan=plan))
            in_progress_plan = load_json(self.get(reassignment_path)[0])
            in_progress_partitions = [
                '{topic}-{p_id}'.format(
                    topic=p_data['topic'],
                    p_id=str(p_data['partition']),
                )
                for p_data in in_progress_plan['partitions']
            ]
            _log.warning(
                '{count} partition(s) reassignment currently in progress:-'
                .format(count=len(in_progress_partitions)),
            )
            _log.warning(
                '{partitions}. In Progress reassignment plan...'.format(
                    partitions=', '.join(in_progress_partitions),
                ),
            )
            return False
        except Exception as e:
            _log.error(
                'Could not re-assign partitions {plan}. Error: {e}'
                .format(plan=plan, e=e),
            )
            return False
Submit reassignment plan for execution.