CompletableFuture<Long> addEntity(String scope, String stream, String entity) {
        // add persistent sequential node to the latest collection number 
        // if collectionNum is sealed, increment collection number and write the entity there. 
        return getLatestCollection(scope, stream)
                .thenCompose(latestcollectionNum ->
                        storeHelper.createPersistentSequentialZNode(getEntitySequentialPath(scope, stream, latestcollectionNum),
                                entity.getBytes(Charsets.UTF_8))
                                   .thenCompose(positionPath -> {
                                       int position = getPositionFromPath(positionPath);
                                       if (position > rollOverAfter) {
                                           // if newly created position exceeds rollover limit, we need to delete that entry 
                                           // and roll over. 
                                           // 1. delete newly created path
                                           return storeHelper.deletePath(positionPath, false)
                                                             // 2. seal latest collection
                                                             .thenCompose(v -> storeHelper.createZNodeIfNotExist(
                                                                     getCollectionSealedPath(scope, stream, latestcollectionNum)))
                                                             // 3. call addEntity recursively
                                                             .thenCompose(v -> addEntity(scope, stream, entity))
                                                             // 4. delete empty sealed collection path
                                                             .thenCompose(orderedPosition -> 
                                                                     tryDeleteSealedCollection(scope, stream, latestcollectionNum)
                                                                     .thenApply(v -> orderedPosition));

                                       } else {
                                           return CompletableFuture.completedFuture(Position.toLong(latestcollectionNum, position));
                                       }
                                   }))
                .whenComplete((r, e) -> {
                    if (e != null) {
                        log.error("error encountered while trying to add entity {} for stream {}/{}", entity, scope, stream, e);
                    } else {
                        log.debug("entity {} added for stream {}/{} at position {}", entity, scope, stream, r);
                    }
                });
    }
CompletableFuture < Long > addEntity ( String scope , String stream , String entity ) { return getLatestCollection ( scope , stream ) . thenCompose ( latestcollectionNum -> storeHelper . createPersistentSequentialZNode ( getEntitySequentialPath ( scope , stream , latestcollectionNum ) , entity . getBytes ( Charsets . UTF_8 ) ) . thenCompose ( positionPath -> { int position = getPositionFromPath ( positionPath ) ; if ( position > rollOverAfter ) { return storeHelper . deletePath ( positionPath , false ) . thenCompose ( v -> storeHelper . createZNodeIfNotExist ( getCollectionSealedPath ( scope , stream , latestcollectionNum ) ) ) . thenCompose ( v -> addEntity ( scope , stream , entity ) ) . thenCompose ( orderedPosition -> tryDeleteSealedCollection ( scope , stream , latestcollectionNum ) . thenApply ( v -> orderedPosition ) ) ; } else { return CompletableFuture . completedFuture ( Position . toLong ( latestcollectionNum , position ) ) ; } } ) ) . whenComplete ( ( r , e ) -> { if ( e != null ) { log . error ( "error encountered while trying to add entity {} for stream {}/{}" , entity , scope , stream , e ) ; } else { log . debug ( "entity {} added for stream {}/{} at position {}" , entity , scope , stream , r ) ; } } ) ; }
Method to add new entity to the ordered collection. Note: Same entity could be added to the collection multiple times.
Entities are added to the latest collection for the stream. If the collection has exhausted positions allowed for by rollOver limit,
a new successor collection is initiated and the entry is added to the new collection.
Any entries with positions higher than rollover values under a set are ignored and eventually purged.

@param scope scope
@param stream stream
@param entity entity to add
@return CompletableFuture which when completed returns the position where the entity is added to the set.