@Override
    public CompletableFuture<Object> updateGlobalAggregate(
            String aggregateName, Object aggregand, byte[] serializedAggregateFunction) {

        AggregateFunction aggregateFunction = null;
        try {
            aggregateFunction =
                    InstantiationUtil.deserializeObject(
                            serializedAggregateFunction, userCodeLoader);
        } catch (Exception e) {
            log.error("Error while attempting to deserialize user AggregateFunction.");
            return FutureUtils.completedExceptionally(e);
        }

        Object accumulator = accumulators.get(aggregateName);
        if (null == accumulator) {
            accumulator = aggregateFunction.createAccumulator();
        }
        accumulator = aggregateFunction.add(aggregand, accumulator);
        accumulators.put(aggregateName, accumulator);
        return CompletableFuture.completedFuture(aggregateFunction.getResult(accumulator));
    }
@Test
    void testJobMasterAggregatesValuesCorrectly() throws Exception {
        try (final JobMaster jobMaster =
                new JobMasterBuilder(jobGraph, rpcService)
                        .withConfiguration(configuration)
                        .withHighAvailabilityServices(haServices)
                        .withHeartbeatServices(heartbeatServices)
                        .createJobMaster()) {

            jobMaster.start();
            final JobMasterGateway jobMasterGateway =
                    jobMaster.getSelfGateway(JobMasterGateway.class);

            CompletableFuture<Object> updateAggregateFuture;

            AggregateFunction<Integer, Integer, Integer> aggregateFunction =
                    createAggregateFunction();

            ClosureCleaner.clean(
                    aggregateFunction, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
            byte[] serializedAggregateFunction =
                    InstantiationUtil.serializeObject(aggregateFunction);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg1", 1, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(1);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg1", 2, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(3);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg1", 3, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(6);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg1", 4, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(10);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg2", 10, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(10);

            updateAggregateFuture =
                    jobMasterGateway.updateGlobalAggregate("agg2", 23, serializedAggregateFunction);
            assertThat(updateAggregateFuture.get()).isEqualTo(33);
        }
    }