Publish AWS Step Function per-activity backlog as Cloudwatch metric

ScheduledEventMetrics.java

    public void metricPendingActivities() {
        AWSStepFunctions client = AWSStepFunctionsClientBuilder.defaultClient();

        ListStateMachinesRequest lsmr = new ListStateMachinesRequest();
        //TODO: paging
        ListStateMachinesResult machinesResult = client.listStateMachines(lsmr);

        Map<String, Long> counts = machinesResult.getStateMachines()
                .stream()
                .map(m -> {
                    ListExecutionsRequest executionsReq = new ListExecutionsRequest()
                            .withMaxResults(100)
                            .withStateMachineArn(m.getStateMachineArn())
                            .withStatusFilter(ExecutionStatus.RUNNING);
                    ListExecutionsResult res = client.listExecutions(executionsReq);
                    return res.getExecutions();
                })
                .flatMap(l -> l.stream())
                .map(e -> {
                    GetExecutionHistoryRequest req = new GetExecutionHistoryRequest()
                            .withExecutionArn(e.getExecutionArn())
                            .withReverseOrder(true);
                    GetExecutionHistoryResult res = client.getExecutionHistory(req);
                    return res.getEvents();
                })
                .map(h -> h.get(0))
                .filter(h -> h.getActivityScheduledEventDetails() != null)
                .map(h -> h.getActivityScheduledEventDetails())
                .collect(Collectors.groupingBy(
                        ActivityScheduledEventDetails::getResource,
                        Collectors.counting()));

        ListActivitiesRequest actr = new ListActivitiesRequest();
        client.listActivities(actr).getActivities().forEach(a -> {
            if(!counts.containsKey(a.getActivityArn()))
                counts.put(a.getActivityArn(), 0L);
        });

        AmazonCloudWatch cw = AmazonCloudWatchClientBuilder.defaultClient();
        counts.entrySet().forEach(e -> {
            PutMetricDataRequest req = new PutMetricDataRequest()
                    .withNamespace("MyCompany/Custom")
                    .withMetricData(new MetricDatum()
                            .withMetricName("PendingTasks")
                            .withValue(e.getValue().doubleValue())
                            .withDimensions(new Dimension()
                                    .withName("ActivityArn")
                                    .withValue(e.getKey())));
            cw.putMetricData(req);
        });

    }

This won't work for machines with parallel branches however