<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def get_reduced_assignment( self, original_assignment, cluster_topology, max_partition_movements, max_leader_only_changes, max_movement_size=DEFAULT_MAX_MOVEMENT_SIZE, force_progress=False, ):
 """Reduce the assignment based on the total actions. Actions represent actual partition movements and/or changes in preferred leader. Get the difference of original and proposed assignment and take the subset of this plan for given limit. Argument(s):
 original_assignment: Current assignment of cluster in zookeeper cluster_topology: Cluster topology containing the new proposed-assignment of cluster max_partition_movements:Maximum number of partition-movements in final set of actions max_leader_only_changes:Maximum number of actions with leader only changes max_movement_size: Maximum size, in bytes, to move in final set of actions force_progress: Whether to force progress if max_movement_size is too small :return: :reduced_assignment: Final reduced assignment """

        new_assignment = cluster_topology.assignment
        if (not original_assignment or not new_assignment or
                max_partition_movements < 0 or max_leader_only_changes < 0 or
                max_movement_size < 0):
            return {}

        # The replica set stays the same for leaders only changes
        leaders_changes = [
            (t_p, new_assignment[t_p])
            for t_p, replica in six.iteritems(original_assignment)
            if replica != new_assignment[t_p] and
            set(replica) == set(new_assignment[t_p])
        ]

        # The replica set is different for partitions changes
        # Here we create a list of tuple ((topic, partion), # replica movements)
        partition_change_count = [
            (
                t_p,
                len(set(replica) - set(new_assignment[t_p])),
            )
            for t_p, replica in six.iteritems(original_assignment)
            if set(replica) != set(new_assignment[t_p])
        ]

        self.log.info(
            "Total number of actions before reduction: %s.",
            len(partition_change_count) + len(leaders_changes),
        )
        # Extract reduced plan maximizing uniqueness of topics and ensuring we do not
        # go over the max_movement_size
        reduced_actions = self._extract_actions_unique_topics(
            partition_change_count,
            max_partition_movements,
            cluster_topology,
            max_movement_size,
        )

        # Ensure progress is made if force_progress=True
        if len(reduced_actions) == 0 and force_progress:
            smallest_size = min([cluster_topology.partitions[t_p[0]].size for t_p in partition_change_count])
            self.log.warning(
                '--max-movement-size={max_movement_size} is too small, using smallest size'
                ' in set of partitions to move, {smallest_size} instead to force progress'.format(
                    max_movement_size=max_movement_size,
                    smallest_size=smallest_size,
                )
            )
            max_movement_size = smallest_size
            reduced_actions = self._extract_actions_unique_topics(
                partition_change_count,
                max_partition_movements,
                cluster_topology,
                max_movement_size,
            )

        reduced_partition_changes = [
            (t_p, new_assignment[t_p]) for t_p in reduced_actions
        ]
        self.log.info(
            "Number of partition changes: %s."
            " Number of leader-only changes: %s",
            len(reduced_partition_changes),
            min(max_leader_only_changes, len(leaders_changes)),
        )
        # Merge leaders and partition changes and generate the assignment
        reduced_assignment = {
            t_p: replicas
            for t_p, replicas in (
                reduced_partition_changes + leaders_changes[:max_leader_only_changes]
            )
        }
        return reduced_assignment