@Override
    public boolean shouldRescale(
            VertexParallelism currentParallelism, VertexParallelism newParallelism) {
        for (JobVertexID vertex : currentParallelism.getVertices()) {
            int parallelismChange =
                    newParallelism.getParallelism(vertex)
                            - currentParallelism.getParallelism(vertex);
            if (parallelismChange != 0) {
                return true;
            }
        }
        return false;
    }
@Test
    void testAlwaysScaleDown() {
        final RescalingController rescalingController =
                new EnforceParallelismChangeRescalingController();
        assertThat(rescalingController.shouldRescale(forParallelism(2), forParallelism(1)))
                .isTrue();
    }