public static Function.FunctionMetaData changeFunctionInstanceStatus(Function.FunctionMetaData functionMetaData,
                                                                         Integer instanceId, boolean start) {
        Function.FunctionMetaData.Builder builder = functionMetaData.toBuilder()
                .setVersion(functionMetaData.getVersion() + 1);
        if (builder.getInstanceStatesMap() == null || builder.getInstanceStatesMap().isEmpty()) {
            for (int i = 0; i < functionMetaData.getFunctionDetails().getParallelism(); ++i) {
                builder.putInstanceStates(i, Function.FunctionState.RUNNING);
            }
        }
        Function.FunctionState state = start ? Function.FunctionState.RUNNING : Function.FunctionState.STOPPED;
        if (instanceId < 0) {
            for (int i = 0; i < functionMetaData.getFunctionDetails().getParallelism(); ++i) {
                builder.putInstanceStates(i, state);
            }
        } else if (instanceId < builder.getFunctionDetails().getParallelism()) {
            builder.putInstanceStates(instanceId, state);
        }
        return builder.build();
    }
@Test
    public void testChangeState() {
        long version = 5;
        Function.FunctionMetaData metaData = Function.FunctionMetaData.newBuilder().setFunctionDetails(
                Function.FunctionDetails.newBuilder().setName("func-1").setParallelism(2)).setVersion(version).build();
        Function.FunctionMetaData newMetaData = FunctionMetaDataUtils.changeFunctionInstanceStatus(metaData, 0, false);
        Assert.assertTrue(newMetaData.getInstanceStatesMap() != null);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().size(), 2);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(0), Function.FunctionState.STOPPED);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(1), Function.FunctionState.RUNNING);
        Assert.assertEquals(newMetaData.getVersion(), version + 1);

        // Nothing should happen
        newMetaData = FunctionMetaDataUtils.changeFunctionInstanceStatus(newMetaData, 3, false);
        Assert.assertTrue(newMetaData.getInstanceStatesMap() != null);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().size(), 2);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(0), Function.FunctionState.STOPPED);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(1), Function.FunctionState.RUNNING);
        Assert.assertEquals(newMetaData.getVersion(), version + 2);

        // Change one more
        newMetaData = FunctionMetaDataUtils.changeFunctionInstanceStatus(newMetaData, 1, false);
        Assert.assertTrue(newMetaData.getInstanceStatesMap() != null);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().size(), 2);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(0), Function.FunctionState.STOPPED);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(1), Function.FunctionState.STOPPED);
        Assert.assertEquals(newMetaData.getVersion(), version + 3);

        // Change all more
        newMetaData = FunctionMetaDataUtils.changeFunctionInstanceStatus(newMetaData, -1, true);
        Assert.assertTrue(newMetaData.getInstanceStatesMap() != null);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().size(), 2);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(0), Function.FunctionState.RUNNING);
        Assert.assertEquals(newMetaData.getInstanceStatesMap().get(1), Function.FunctionState.RUNNING);
        Assert.assertEquals(newMetaData.getVersion(), version + 4);
    }