public static boolean taskConfigsChanged(ClusterConfigState configState, String connName, List<Map<String, String>> rawTaskProps) {
        int currentNumTasks = configState.taskCount(connName);
        boolean result = false;
        if (rawTaskProps.size() != currentNumTasks) {
            log.debug("Connector {} task count changed from {} to {}", connName, currentNumTasks, rawTaskProps.size());
            result = true;
        }
        if (!result) {
            for (int index = 0; index < currentNumTasks; index++) {
                ConnectorTaskId taskId = new ConnectorTaskId(connName, index);
                if (!rawTaskProps.get(index).equals(configState.rawTaskConfig(taskId))) {
                    log.debug("Connector {} has change in configuration for task {}-{}", connName, connName, index);
                    result = true;
                }
            }
        }
        if (!result) {
            Map<String, String> appliedConnectorConfig = configState.appliedConnectorConfig(connName);
            Map<String, String> currentConnectorConfig = configState.connectorConfig(connName);
            if (!Objects.equals(appliedConnectorConfig, currentConnectorConfig)) {
                log.debug("Forcing task restart for connector {} as its configuration appears to be updated", connName);
                result = true;
            }
        }
        if (result) {
            log.debug("Reconfiguring connector {}: writing new updated configurations for tasks", connName);
        } else {
            log.debug("Skipping reconfiguration of connector {} as generated configs appear unchanged", connName);
        }
        return result;
    }
@Test
    public void testTaskConfigComparison() {
        ClusterConfigState snapshot = mock(ClusterConfigState.class);

        when(snapshot.taskCount(CONN1)).thenReturn(TASK_CONFIGS.size());
        TASK_CONFIGS_MAP.forEach((task, config) -> when(snapshot.rawTaskConfig(task)).thenReturn(config));
        // Same task configs, same number of tasks--no change
        assertFalse(AbstractHerder.taskConfigsChanged(snapshot, CONN1, TASK_CONFIGS));

        when(snapshot.taskCount(CONN1)).thenReturn(TASK_CONFIGS.size() + 1);
        // Different number of tasks; should report a change
        assertTrue(AbstractHerder.taskConfigsChanged(snapshot, CONN1, TASK_CONFIGS));

        when(snapshot.taskCount(CONN1)).thenReturn(TASK_CONFIG.size());
        List<Map<String, String>> alteredTaskConfigs = new ArrayList<>(TASK_CONFIGS);
        alteredTaskConfigs.set(alteredTaskConfigs.size() - 1, Collections.emptyMap());
        // Last task config is different; should report a change
        assertTrue(AbstractHerder.taskConfigsChanged(snapshot, CONN1, alteredTaskConfigs));

        // Make sure we used exclusively raw task configs and never attempted transformation,
        // since otherwise failures in transformation could either cause an infinite loop of task
        // config generation, or prevent any task configs from being published
        verify(snapshot, never()).taskConfig(any());
    }