public static boolean canChangeState(Function.FunctionMetaData functionMetaData, int instanceId,
                                         Function.FunctionState newState) {
        if (instanceId >= functionMetaData.getFunctionDetails().getParallelism()) {
            return false;
        }
        if (functionMetaData.getInstanceStatesMap() == null || functionMetaData.getInstanceStatesMap().isEmpty()) {
            // This means that all instances of the functions are running
            return newState == Function.FunctionState.STOPPED;
        }
        if (instanceId >= 0) {
            if (functionMetaData.getInstanceStatesMap().containsKey(instanceId)) {
                return functionMetaData.getInstanceStatesMap().get(instanceId) != newState;
            } else {
                return false;
            }
        } else {
            // want to change state for all instances
            for (Function.FunctionState state : functionMetaData.getInstanceStatesMap().values()) {
                if (state != newState) {
                    return true;
                }
            }
            return false;
        }
    }
@Test
    public void testCanChangeState() {

        long version = 5;
        Function.FunctionMetaData metaData = Function.FunctionMetaData.newBuilder().setFunctionDetails(
                Function.FunctionDetails.newBuilder().setName("func-1").setParallelism(2)).setVersion(version).build();

        Assert.assertTrue(FunctionMetaDataUtils.canChangeState(metaData, 0, Function.FunctionState.STOPPED));
        Assert.assertFalse(FunctionMetaDataUtils.canChangeState(metaData, 0, Function.FunctionState.RUNNING));
        Assert.assertFalse(FunctionMetaDataUtils.canChangeState(metaData, 2, Function.FunctionState.STOPPED));
        Assert.assertFalse(FunctionMetaDataUtils.canChangeState(metaData, 2, Function.FunctionState.RUNNING));
    }