Map < WorkflowInstance , RunState > readActiveStates ( ) throws IOException { var timeout = CompletableFuture . runAsync ( ( ) -> { } , delayedExecutor ( 30 , SECONDS ) ) ; var instances = listActiveInstances0 ( timeout ) ; // Strongly consistently read values for the instances in parallel var states = gatherIO ( Lists . partition ( List . copyOf ( instances ) , MAX_NUMBER_OF_ENTITIES_IN_ONE_BATCH_READ ) . stream ( ) . map ( batch -> asyncIO ( ( ) -> readRunStateBatch ( batch ) ) ) . collect ( toList ( ) ) , timeout ) . stream ( ) . flatMap ( Collection :: stream ) . collect ( toMap ( RunState :: workflowInstance , Function . identity ( ) ) ) ; timeout . cancel ( true ) ; return states ; }
Strongly consistently read all active states