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 testTaskConfigsChangedWhenAppliedConnectorConfigDiffers() {
        assertFalse(AbstractHerder.taskConfigsChanged(SNAPSHOT, CONN1, TASK_CONFIGS));

        ClusterConfigState snapshotWithNoAppliedConfig = new ClusterConfigState(
                1,
                null,
                Collections.singletonMap(CONN1, 3),
                Collections.singletonMap(CONN1, CONN1_CONFIG),
                Collections.singletonMap(CONN1, TargetState.STARTED),
                TASK_CONFIGS_MAP,
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptySet(),
                Collections.emptySet()
        );
        assertTrue(AbstractHerder.taskConfigsChanged(snapshotWithNoAppliedConfig, CONN1, TASK_CONFIGS));

        Map<String, String> appliedConfig = new HashMap<>(CONN1_CONFIG);
        String newTopicsProperty = appliedConfig.getOrDefault(SinkConnectorConfig.TOPICS_CONFIG, "foo") + ",newTopic";
        appliedConfig.put(SinkConnectorConfig.TOPICS_CONFIG, newTopicsProperty);
        ClusterConfigState snapshotWithDifferentAppliedConfig = new ClusterConfigState(
                1,
                null,
                Collections.singletonMap(CONN1, 3),
                Collections.singletonMap(CONN1, CONN1_CONFIG),
                Collections.singletonMap(CONN1, TargetState.STARTED),
                TASK_CONFIGS_MAP,
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.singletonMap(CONN1, new AppliedConnectorConfig(appliedConfig)),
                Collections.emptySet(),
                Collections.emptySet()
        );
        assertTrue(AbstractHerder.taskConfigsChanged(snapshotWithDifferentAppliedConfig, CONN1, TASK_CONFIGS));
    }